From a4470ea6effbff65a276670102c91f1034793739 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 24 Sep 2026 14:00:30 +0000 Subject: [PATCH] feat(router): configurable relay circuit limits The router relay used go-libp2p's defaults, cutting every relayed connection after 2 minutes or 128 KiB per direction. Add --relay-limit-duration (default 1h) and --relay-limit-data (default no limit, accepts sizes like 512MiB) to sam-router, and the matching --router-relay-limit-* tunables to sam-one. 0 means no limit. go.mod: go-humanize and golang.org/x/sys become direct dependencies (the latter was already imported by the mobile FFI). Part of #483 --- cmd/sam-one/main.go | 2 + cmd/sam-router/main.go | 6 ++ go.mod | 4 +- internal/router/config.go | 4 + internal/router/relay_limit.go | 63 +++++++++++ internal/router/relay_limit_test.go | 141 +++++++++++++++++++++++++ internal/router/router.go | 3 +- internal/standalone/standalone.go | 8 ++ site/content/docs/reference/router.md | 1 + site/content/docs/reference/sam-one.md | 1 + 10 files changed, 230 insertions(+), 3 deletions(-) create mode 100644 internal/router/relay_limit.go create mode 100644 internal/router/relay_limit_test.go diff --git a/cmd/sam-one/main.go b/cmd/sam-one/main.go index 8a859f78..fedb012e 100644 --- a/cmd/sam-one/main.go +++ b/cmd/sam-one/main.go @@ -222,6 +222,8 @@ func main() { rootCmd.Flags().IntVar(&routerTunables.ConnsPerSourceIP, "router-conns-per-source-ip", 0, "Per-source-IP connection budget (0 follows the high watermark; proxied peers share source IPs)") rootCmd.Flags().DurationVar(&routerTunables.DHTProviderAddrTTL, "router-dht-provider-addr-ttl", 0, "DHT provider address TTL (0 keeps the library default)") rootCmd.Flags().DurationVar(&routerTunables.DHTMaxRecordAge, "router-dht-max-record-age", 0, "DHT record max age (0 keeps the library default)") + rootCmd.Flags().DurationVar(&routerTunables.RelayLimitDuration, "router-relay-limit-duration", 0, "Relayed connection lifetime (0 keeps the component default of 1h; use e.g. 24h for longer)") + rootCmd.Flags().Var(&routerTunables.RelayLimitData, "router-relay-limit-data", "Bytes relayed per direction per connection, e.g. 512MiB (0 keeps the component default: no limit)") rootCmd.Flags().BoolVar(&routerAllowLoopback, "router-allow-loopback", true, "Advertise loopback addresses (disable on public deployments)") rootCmd.AddCommand(newAdminSubcommands()...) diff --git a/cmd/sam-router/main.go b/cmd/sam-router/main.go index da37ad63..168ea3a0 100644 --- a/cmd/sam-router/main.go +++ b/cmd/sam-router/main.go @@ -46,6 +46,8 @@ var ( lowWaterMark int highWaterMark int metricsAddr string + relayLimitDuration time.Duration + relayLimitData router.ByteSize ) var logger = golog.Logger("sam-router-cli") @@ -86,6 +88,8 @@ func main() { LowWaterMark: lowWaterMark, HighWaterMark: highWaterMark, MetricsAddr: metricsAddr, + RelayLimitDuration: relayLimitDuration, + RelayLimitData: int64(relayLimitData), } r, err := router.NewRouter(cmd.Context(), opts) @@ -125,6 +129,8 @@ func main() { rootCmd.Flags().IntVar(&lowWaterMark, "low-watermark", 1000, "Connection manager low watermark limit") rootCmd.Flags().IntVar(&highWaterMark, "high-watermark", 4000, "Connection manager high watermark limit") rootCmd.Flags().StringVar(&metricsAddr, "metrics-addr", "", "Serve Prometheus /metrics, /healthz and /readyz on this address (e.g. 0.0.0.0:9090); unauthenticated, off by default") + rootCmd.Flags().DurationVar(&relayLimitDuration, "relay-limit-duration", router.DefaultRelayLimitDuration, "Lifetime of each relayed connection (0 = no limit)") + rootCmd.Flags().Var(&relayLimitData, "relay-limit-data", "Bytes relayed per direction on each relayed connection, e.g. 512MiB (0 = no limit)") ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancel() diff --git a/go.mod b/go.mod index 7647e0d4..61afecb0 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/a2aproject/a2a-go/v2 v2.5.0 github.com/biscuit-auth/biscuit-go/v2 v2.2.0 github.com/coreos/go-oidc/v3 v3.21.0 + github.com/dustin/go-humanize v1.0.1 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/ipfs/go-cid v0.6.2 @@ -31,6 +32,7 @@ require ( go.uber.org/zap v1.28.0 golang.org/x/net v0.59.0 golang.org/x/oauth2 v0.37.0 + golang.org/x/sys v0.48.0 golang.org/x/time v0.16.0 google.golang.org/protobuf v1.36.12 gopkg.in/yaml.v2 v2.4.0 @@ -47,7 +49,6 @@ require ( github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 // indirect github.com/dunglas/httpsfv v1.1.0 // indirect - github.com/dustin/go-humanize v1.0.1 // indirect github.com/filecoin-project/go-clock v0.1.0 // indirect github.com/flynn/noise v1.1.0 // indirect github.com/go-jose/go-jose/v4 v4.1.4 // indirect @@ -141,7 +142,6 @@ require ( golang.org/x/exp v0.0.0-20260718201538-764159d718ef // indirect golang.org/x/mod v0.41.0 // indirect golang.org/x/sync v0.23.0 // indirect - golang.org/x/sys v0.48.0 // indirect golang.org/x/telemetry v0.0.0-20260811182544-a038080d80e5 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/tools v0.49.0 // indirect diff --git a/internal/router/config.go b/internal/router/config.go index 58cf29cd..21e7a2f5 100644 --- a/internal/router/config.go +++ b/internal/router/config.go @@ -43,6 +43,10 @@ type Options struct { DHTMaxRecordAge time.Duration LowWaterMark int HighWaterMark int + // RelayLimitDuration / RelayLimitData cap each relayed connection; 0 is + // no limit, so Default() leaves them alone. + RelayLimitDuration time.Duration + RelayLimitData int64 // RequiredRole restricts enrollment and startup to only accept tokens containing this role. RequiredRole string // HTTPFallbackHandler, when set, serves ordinary (non-WebSocket-upgrade) diff --git a/internal/router/relay_limit.go b/internal/router/relay_limit.go new file mode 100644 index 00000000..49c1dce5 --- /dev/null +++ b/internal/router/relay_limit.go @@ -0,0 +1,63 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package router + +import ( + "fmt" + "math" + "time" + + "github.com/dustin/go-humanize" + "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/relay" +) + +// DefaultRelayLimitDuration is how long a relayed connection lives unless the operator sets it. +const DefaultRelayLimitDuration = time.Hour + +// relayLimit maps operator limits (0 = none) onto go-libp2p, where only a nil +// limit is unlimited and a zero field cuts every circuit at once. +func relayLimit(duration time.Duration, data int64) *relay.RelayLimit { + if duration <= 0 && data <= 0 { + return nil + } + limit := &relay.RelayLimit{Duration: duration, Data: data} + if duration <= 0 { + // The relay advertises seconds as uint32; anything larger wraps. + limit.Duration = math.MaxUint32 * time.Second + } + if data <= 0 { + limit.Data = math.MaxInt64 + } + return limit +} + +// ByteSize is a flag value that accepts human sizes such as 128MiB. +type ByteSize int64 + +func (b *ByteSize) String() string { return humanize.IBytes(uint64(*b)) } + +func (b *ByteSize) Type() string { return "size" } + +func (b *ByteSize) Set(s string) error { + n, err := humanize.ParseBytes(s) + if err != nil { + return err + } + if n > math.MaxInt64 { + return fmt.Errorf("size %q is too large", s) + } + *b = ByteSize(n) + return nil +} diff --git a/internal/router/relay_limit_test.go b/internal/router/relay_limit_test.go new file mode 100644 index 00000000..817ad428 --- /dev/null +++ b/internal/router/relay_limit_test.go @@ -0,0 +1,141 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package router + +import ( + "context" + "io" + "math" + "reflect" + "testing" + "time" + + "github.com/libp2p/go-libp2p" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/client" + "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/relay" + "github.com/multiformats/go-multiaddr" +) + +func TestRelayLimit(t *testing.T) { + unlimitedDuration := math.MaxUint32 * time.Second + for _, tc := range []struct { + name string + duration time.Duration + data int64 + want *relay.RelayLimit + }{ + {"no limits", 0, 0, nil}, + {"duration only", time.Hour, 0, &relay.RelayLimit{Duration: time.Hour, Data: math.MaxInt64}}, + {"data only", 0, 4096, &relay.RelayLimit{Duration: unlimitedDuration, Data: 4096}}, + {"both", time.Minute, 4096, &relay.RelayLimit{Duration: time.Minute, Data: 4096}}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := relayLimit(tc.duration, tc.data); !reflect.DeepEqual(got, tc.want) { + t.Errorf("relayLimit(%v, %d) = %+v, want %+v", tc.duration, tc.data, got, tc.want) + } + }) + } +} + +func TestByteSizeSet(t *testing.T) { + for _, tc := range []struct { + in string + want ByteSize + wantErr bool + }{ + {"128MiB", 128 << 20, false}, + {"1GB", 1_000_000_000, false}, + {"0", 0, false}, + {"abc", 0, true}, + {"20EiB", 0, true}, + } { + var got ByteSize + err := got.Set(tc.in) + if (err != nil) != tc.wantErr || got != tc.want { + t.Errorf("Set(%q) = %d, %v; want %d, error %v", tc.in, got, err, tc.want, tc.wantErr) + } + } +} + +// A real circuit: the data cap truncates, and a duration-only limit must not cut. +func TestRelayLimitOnCircuit(t *testing.T) { + const payload = 8192 + for _, tc := range []struct { + name string + limit *relay.RelayLimit + complete bool + }{ + {"data cap cuts the circuit", relayLimit(0, 4096), false}, + {"duration only passes everything", relayLimit(time.Hour, 0), true}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + relayHost := newLoopbackHost(t, libp2p.DisableRelay()) + if _, err := relay.New(relayHost, relay.WithLimit(tc.limit)); err != nil { + t.Fatal(err) + } + relayInfo := peer.AddrInfo{ID: relayHost.ID(), Addrs: relayHost.Addrs()} + + dst := newLoopbackHost(t, libp2p.EnableRelay()) + dst.SetStreamHandler("/test", func(s network.Stream) { + _, _ = s.Write(make([]byte, payload)) + _ = s.Close() + }) + if err := dst.Connect(ctx, relayInfo); err != nil { + t.Fatal(err) + } + if _, err := client.Reserve(ctx, dst, relayInfo); err != nil { + t.Fatal(err) + } + + src := newLoopbackHost(t, libp2p.EnableRelay()) + if err := src.Connect(ctx, relayInfo); err != nil { + t.Fatal(err) + } + circuit, err := multiaddr.NewMultiaddr("/p2p/" + relayHost.ID().String() + "/p2p-circuit") + if err != nil { + t.Fatal(err) + } + if err := src.Connect(ctx, peer.AddrInfo{ID: dst.ID(), Addrs: []multiaddr.Multiaddr{relayHost.Addrs()[0].Encapsulate(circuit)}}); err != nil { + t.Fatal(err) + } + + s, err := src.NewStream(network.WithAllowLimitedConn(ctx, "test"), dst.ID(), "/test") + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + got, _ := io.Copy(io.Discard, s) + if (got == payload) != tc.complete { + t.Errorf("read %d of %d bytes through the relay, want complete=%v", got, payload, tc.complete) + } + }) + } +} + +func newLoopbackHost(t *testing.T, opts ...libp2p.Option) host.Host { + t.Helper() + h, err := libp2p.New(append(opts, libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"))...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = h.Close() }) + return h +} diff --git a/internal/router/router.go b/internal/router/router.go index 87d35778..36b571b7 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -407,7 +407,8 @@ func (r *Router) Start() error { } // Setup Relay - _, err = relay.New(hostNode, relay.WithACL(&relayACL{r: r})) + _, err = relay.New(hostNode, relay.WithACL(&relayACL{r: r}), + relay.WithLimit(relayLimit(r.config.RelayLimitDuration, r.config.RelayLimitData))) if err != nil { _ = hostNode.Close() return err diff --git a/internal/standalone/standalone.go b/internal/standalone/standalone.go index 9bcc8ee6..7f37b107 100644 --- a/internal/standalone/standalone.go +++ b/internal/standalone/standalone.go @@ -147,6 +147,9 @@ type RouterTunables struct { // DHTProviderAddrTTL / DHTMaxRecordAge tune DHT record lifetimes. DHTProviderAddrTTL time.Duration DHTMaxRecordAge time.Duration + // RelayLimitDuration / RelayLimitData cap each relayed connection. + RelayLimitDuration time.Duration + RelayLimitData router.ByteSize // DisallowLoopback stops advertising loopback addresses (useful on // public deployments; the default keeps local development working). DisallowLoopback bool @@ -177,6 +180,9 @@ func (o *Options) Default() { if o.Router.ConnsPerSourceIP == 0 { o.Router.ConnsPerSourceIP = o.Router.HighWaterMark } + if o.Router.RelayLimitDuration == 0 { + o.Router.RelayLimitDuration = router.DefaultRelayLimitDuration + } } // Validate rejects option combinations Start could not honor. @@ -342,6 +348,8 @@ func (s *Server) Start(ctx context.Context) error { HighWaterMark: s.opts.Router.HighWaterMark, DHTProviderAddrTTL: s.opts.Router.DHTProviderAddrTTL, DHTMaxRecordAge: s.opts.Router.DHTMaxRecordAge, + RelayLimitDuration: s.opts.Router.RelayLimitDuration, + RelayLimitData: int64(s.opts.Router.RelayLimitData), // Single-port deployments typically sit behind a TLS-terminating // proxy (Cloud Run, L7 LBs) or NAT where every peer shares a few // source IPs; libp2p's default 8-conns-per-IP cap would throttle diff --git a/site/content/docs/reference/router.md b/site/content/docs/reference/router.md index 15026c54..7fd8a2d6 100644 --- a/site/content/docs/reference/router.md +++ b/site/content/docs/reference/router.md @@ -43,6 +43,7 @@ replica count. | `--conns-per-source-ip` | `8` (libp2p default) | Inbound connections accepted per source IP. Raise it behind a TLS-terminating proxy or a NAT that puts many peers on one address. | | `--low-watermark`, `--high-watermark` | `1000`, `4000` | Connection manager limits. Above the high mark, connections are trimmed down to the low mark. | | `--dht-provider-addr-ttl`, `--dht-max-record-age` | library defaults | DHT record lifetimes. | +| `--relay-limit-duration`, `--relay-limit-data` | `1h`, `0` | Caps on each relayed connection: lifetime, and bytes per direction (`512MiB`, `1GB`). The relay cuts the connection when either is reached. `0` means no limit. | | `--metrics-addr` | off | Serve `/metrics`, `/healthz` and `/readyz` without authentication on this address. `/readyz` returns `200` once the router is enrolled and the libp2p host is up. Keep this address separate from the libp2p ports and inside the cluster. | | `--log-level` | `info` | `debug`, `info`, `warn`, `error`. `LOG_FORMAT=json` selects JSON output. | diff --git a/site/content/docs/reference/sam-one.md b/site/content/docs/reference/sam-one.md index cf52425e..7d222b31 100644 --- a/site/content/docs/reference/sam-one.md +++ b/site/content/docs/reference/sam-one.md @@ -63,6 +63,7 @@ of zero leaves the default of that component unchanged. | `--router-low-watermark`, `--router-high-watermark` | `--low-watermark`, `--high-watermark` | | `--router-conns-per-source-ip` | `--conns-per-source-ip`. `0` follows the high watermark, because peers behind a proxy share source IPs. | | `--router-dht-provider-addr-ttl`, `--router-dht-max-record-age` | The DHT record lifetimes. | +| `--router-relay-limit-duration`, `--router-relay-limit-data` | `--relay-limit-duration`, `--relay-limit-data`. `0` keeps `1h` and no data limit, so set a long duration such as `24h` instead of an unlimited one. | | `--router-allow-loopback` | `--allow-loopback`. Defaults to `true` here, for a laptop. Disable it on a public deployment. | ## The banner