amadan
relay: bound refusal state, meter rejected per spec, exempt STUN control from buckets, close orphaned allocations (review fix)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 files changed,
+497
−73
go.mod+1 −1internal/relay/server.go+303 −40internal/relay/server_test.go+193 −32
diff --git a/go.mod b/go.mod| index 0d683e268..c7c41ef8f 100644 |
| --- a/go.mod |
| +++ b/go.mod |
| @@ -19,6 +19,7 @@ require ( |
| github.com/aws/smithy-go v1.28.1 |
| github.com/benbjohnson/litestream v0.5.14 |
| github.com/pion/logging v0.2.4 |
| + github.com/pion/stun/v3 v3.0.1 |
| github.com/pion/turn/v4 v4.1.4 |
| golang.org/x/crypto v0.52.0 |
| golang.org/x/net v0.55.0 |
| @@ -57,7 +58,6 @@ require ( |
| github.com/pierrec/lz4/v4 v4.1.22 // indirect |
| github.com/pion/dtls/v3 v3.0.7 // indirect |
| github.com/pion/randutil v0.1.0 // indirect |
| - github.com/pion/stun/v3 v3.0.1 // indirect |
| github.com/pion/transport/v3 v3.0.8 // indirect |
| github.com/pion/transport/v4 v4.0.1 // indirect |
| github.com/prometheus/client_golang v1.17.0 // indirect |
diff --git a/internal/relay/server.go b/internal/relay/server.go| index 403d01380..4e9bc4d48 100644 |
| --- a/internal/relay/server.go |
| +++ b/internal/relay/server.go |
| @@ -1,6 +1,7 @@ |
| package relay |
| import ( |
| + "encoding/binary" |
| "errors" |
| "fmt" |
| "math/rand/v2" |
| @@ -11,6 +12,7 @@ import ( |
| "time" |
| "github.com/pion/logging" |
| + "github.com/pion/stun/v3" |
| "github.com/pion/turn/v4" |
| ) |
| @@ -28,6 +30,25 @@ const pendingLife = 30 * time.Second |
| // and never allocates cannot grow the map without bound. |
| const authMemory = 10 * time.Minute |
| +const ( |
| + // maxRefusalIdentities and maxLoggedRefusals bound the two maps an |
| + // unauthenticated client can otherwise grow: pion calls AuthHandler |
| + // with the raw username before it checks MESSAGE-INTEGRITY, so every |
| + // counter keyed on something the caller chose is remote memory. Only |
| + // identities in the authority record get their own row; everything |
| + // else shares the zero identity, and these caps are the backstop. |
| + maxRefusalIdentities = 4096 |
| + maxLoggedRefusals = 4096 |
| + |
| + // refusalLogEvery is how often one identity's one refusal reason may |
| + // put a line in the log. |
| + refusalLogEvery = time.Minute |
| + |
| + // minLastAuthEntries is the floor under Limits.BoxLive*2 for how many |
| + // typed sources may hold a remembered username at once. |
| + minLastAuthEntries = 256 |
| +) |
| + |
| var errTCPAllocUnsupported = errors.New("relay: TCP relay allocations are not supported") |
| // ServerConfig is everything the relay needs from its process: where to |
| @@ -39,6 +60,7 @@ type ServerConfig struct { |
| UDP, TCP string // listen addresses, e.g. ":3478"; empty disables |
| MinPort, MaxPort uint16 // 49152, 65535 |
| DenyCIDRs []*net.IPNet // from CARLOS_RELAY_DENY_CIDRS, on top of the built-in private ranges |
| + ExtraAllow []*net.IPNet // peer ranges allowed despite the built-in list; empty in production |
| Masters func() *Masters |
| Authority func() *Authority |
| Lease func(now time.Time) bool |
| @@ -61,11 +83,13 @@ type Server struct { |
| now func() time.Time |
| log func(string, ...any) |
| - adm *Admission |
| - box *Bucket |
| - ports *portSet |
| - deny []*net.IPNet |
| - local []net.IP // our listeners' own addresses, when not wildcard |
| + adm *Admission |
| + box *Bucket |
| + ports *portSet |
| + deny []*net.IPNet |
| + allow []*net.IPNet |
| + local []net.IP // our listeners' own addresses, when not wildcard |
| + maxLastAuth int |
| turn *turn.Server |
| udp net.PacketConn |
| @@ -80,23 +104,87 @@ type Server struct { |
| bound map[string]*Binding // typed source -> the identity it is bound to |
| allocs map[string]*allocEntry // typed source -> its live allocation |
| owners map[*relayConn]*allocEntry // socket -> what closing it must undo |
| + byPort map[int]*relayConn // relay port -> its socket, however far the allocation got |
| refusals map[Identity]map[string]int |
| + logged map[logKey]time.Time |
| shedding map[Identity]bool |
| } |
| +type logKey struct { |
| + id Identity |
| + reason string |
| +} |
| + |
| type allocEntry struct { |
| conn *relayConn |
| id Identity |
| key string // the typed source this allocation was created for |
| seq int64 |
| timer *time.Timer |
| + flow *flow |
| } |
| type pendingConn struct { |
| conn *relayConn |
| + flow *flow |
| at time.Time |
| } |
| +// flow is one allocation's peer-to-client accounting. pion's allocation |
| +// reads a peer packet and then either forwards it to the client or drops |
| +// it for want of a permission, both on the allocation's own goroutine, so |
| +// a packet that is still unforwarded when the next one arrives (or when |
| +// the socket closes) is a packet that was dropped. That is what spec §3 |
| +// counts as `rejected`, and it is never charged to a budget. |
| +type flow struct { |
| + mu sync.Mutex |
| + awaiting int64 // size of the peer packet read and not yet forwarded |
| +} |
| + |
| +// read records a peer packet and returns the bytes of the previous one if |
| +// it was never forwarded. |
| +func (f *flow) read(n int) int64 { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + dropped := f.awaiting |
| + f.awaiting = int64(n) |
| + return dropped |
| +} |
| + |
| +// forwarded clears the packet the listener has just written to the client. |
| +func (f *flow) forwarded() { |
| + f.mu.Lock() |
| + f.awaiting = 0 |
| + f.mu.Unlock() |
| +} |
| + |
| +// flush is the last packet's verdict when the socket closes. |
| +func (f *flow) flush() int64 { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + dropped := f.awaiting |
| + f.awaiting = 0 |
| + return dropped |
| +} |
| + |
| +// readCounter is the relay socket's peer-ingress leg. It sits under |
| +// relayConn so that relayConn keeps counting only what it forwards. |
| +type readCounter struct { |
| + net.PacketConn |
| + f *flow |
| + onDrop func(int64) |
| +} |
| + |
| +func (r *readCounter) ReadFrom(p []byte) (int, net.Addr, error) { |
| + n, addr, err := r.PacketConn.ReadFrom(p) |
| + if n > 0 { |
| + if dropped := r.f.read(n); dropped > 0 && r.onDrop != nil { |
| + r.onDrop(dropped) |
| + } |
| + } |
| + return n, addr, err |
| +} |
| + |
| type authRec struct { |
| u Username |
| at time.Time |
| @@ -153,22 +241,30 @@ func NewServer(cfg ServerConfig) (*Server, error) { |
| return nil, errors.New("relay: no listener configured") |
| } |
| + maxLastAuth := cfg.Limits.BoxLive * 2 |
| + if maxLastAuth < minLastAuthEntries { |
| + maxLastAuth = minLastAuthEntries |
| + } |
| s := &Server{ |
| - cfg: cfg, |
| - now: cfg.Now, |
| - log: cfg.Log, |
| - adm: NewAdmission(cfg.Limits, cfg.Now), |
| - box: NewBucket(cfg.Limits.BoxBitsPerSec, cfg.Now), |
| - ports: &portSet{m: map[int]bool{}}, |
| - deny: append(append([]*net.IPNet{}, builtinDeny...), cfg.DenyCIDRs...), |
| - pending: map[int]pendingConn{}, |
| - lastAuth: map[string]authRec{}, |
| - admits: map[string]admitRec{}, |
| - bound: map[string]*Binding{}, |
| - allocs: map[string]*allocEntry{}, |
| - owners: map[*relayConn]*allocEntry{}, |
| - refusals: map[Identity]map[string]int{}, |
| - shedding: map[Identity]bool{}, |
| + cfg: cfg, |
| + now: cfg.Now, |
| + log: cfg.Log, |
| + adm: NewAdmission(cfg.Limits, cfg.Now), |
| + box: NewBucket(cfg.Limits.BoxBitsPerSec, cfg.Now), |
| + ports: &portSet{m: map[int]bool{}}, |
| + deny: append(append([]*net.IPNet{}, builtinDeny...), cfg.DenyCIDRs...), |
| + allow: append([]*net.IPNet{}, cfg.ExtraAllow...), |
| + maxLastAuth: maxLastAuth, |
| + pending: map[int]pendingConn{}, |
| + lastAuth: map[string]authRec{}, |
| + admits: map[string]admitRec{}, |
| + bound: map[string]*Binding{}, |
| + allocs: map[string]*allocEntry{}, |
| + owners: map[*relayConn]*allocEntry{}, |
| + byPort: map[int]*relayConn{}, |
| + refusals: map[Identity]map[string]int{}, |
| + logged: map[logKey]time.Time{}, |
| + shedding: map[Identity]bool{}, |
| } |
| var packets []turn.PacketConnConfig |
| @@ -411,6 +507,21 @@ func (s *Server) tick() { |
| delete(s.lastAuth, key) |
| } |
| } |
| + // Refusal counters live only for apps the authority still names; the |
| + // zero identity is the shared bucket for everything else and stays. |
| + for id := range s.refusals { |
| + if id == (Identity{}) { |
| + continue |
| + } |
| + if _, ok := s.cfg.Authority().Lookup(id.Account, id.App); !ok { |
| + delete(s.refusals, id) |
| + } |
| + } |
| + for k, at := range s.logged { |
| + if now.Sub(at) >= refusalLogEvery { |
| + delete(s.logged, k) |
| + } |
| + } |
| s.mu.Unlock() |
| for _, id := range lapsed { |
| @@ -470,11 +581,17 @@ func (s *Server) AllocatePacketConn(network string, requestedPort int) (net.Pack |
| continue |
| } |
| bucket := NewBucket(s.cfg.Limits.AllocBitsPerSec, s.now) |
| - c := newRelayConn(pc, &Binding{}, s.cfg.RelayIP, s.ports, bucket, s.box, |
| - s.countIn, s.countRejected, func(rc *relayConn) { s.release(port, rc) }) |
| + f := &flow{} |
| + var c *relayConn |
| + // The socket counts peer ingress under the wrapper, so a peer |
| + // packet pion drops for want of a permission is still seen. |
| + counted := &readCounter{PacketConn: pc, f: f, onDrop: func(n int64) { s.countRejected(c.b.ID, n) }} |
| + c = newRelayConn(counted, &Binding{}, s.cfg.RelayIP, s.ports, bucket, s.box, |
| + s.countIn, s.countGated, func(rc *relayConn) { s.release(port, rc) }) |
| s.ports.add(port) |
| s.mu.Lock() |
| - s.pending[port] = pendingConn{conn: c, at: s.now()} |
| + s.pending[port] = pendingConn{conn: c, flow: f, at: s.now()} |
| + s.byPort[port] = c |
| s.mu.Unlock() |
| return c, &net.UDPAddr{IP: s.cfg.RelayIP, Port: port}, nil |
| } |
| @@ -537,12 +654,40 @@ func (s *Server) auth(username, realm string, src net.Addr) ([]byte, bool) { |
| } |
| key := DeriveKey(master.Key, u.Account, u.App, u.Incarnation) |
| - s.mu.Lock() |
| - s.lastAuth[typedSrc(src)] = authRec{u: u, at: now} |
| - s.mu.Unlock() |
| + s.rememberAuth(typedSrc(src), u, now) |
| return turn.GenerateAuthKey(username, realm, Credential(key, username)), true |
| } |
| +// rememberAuth stores the username a source last authenticated with, for |
| +// the allocation event that follows. pion verifies MESSAGE-INTEGRITY only |
| +// after the handler returns, so this map is written for sources that have |
| +// not proved anything yet: tick ages entries out, and this caps them at |
| +// twice the box's live-allocation limit, dropping the oldest quarter when |
| +// it fills. |
| +func (s *Server) rememberAuth(key string, u Username, now time.Time) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if _, known := s.lastAuth[key]; !known && len(s.lastAuth) >= s.maxLastAuth { |
| + type aged struct { |
| + key string |
| + at time.Time |
| + } |
| + all := make([]aged, 0, len(s.lastAuth)) |
| + for k, r := range s.lastAuth { |
| + all = append(all, aged{k, r.at}) |
| + } |
| + sort.Slice(all, func(i, j int) bool { return all[i].at.Before(all[j].at) }) |
| + drop := len(all) / 4 |
| + if drop == 0 { |
| + drop = 1 |
| + } |
| + for _, a := range all[:drop] { |
| + delete(s.lastAuth, a.key) |
| + } |
| + } |
| + s.lastAuth[key] = authRec{u: u, at: now} |
| +} |
| + |
| // quota is pion's QuotaHandler, called on Allocate only: a refusal answers |
| // 486. The lease comes first, because a relay whose authority read is |
| // stale must take nothing new on. |
| @@ -591,8 +736,17 @@ func (s *Server) created(src, _ net.Addr, _, _, _ string, relayAddr net.Addr, _ |
| s.mu.Lock() |
| p, havePending := s.pending[ua.Port] |
| if !havePending { |
| + // The sweep already reclaimed this socket, or it was never ours. |
| + // Either way pion now owns an allocation no deadline of ours can |
| + // end, so end it here by closing the socket it reads from. |
| + orphan := s.byPort[ua.Port] |
| s.mu.Unlock() |
| - s.log("relay: allocation created on port %d with no socket of ours", ua.Port) |
| + if orphan != nil { |
| + s.log("relay: allocation created on port %d after its socket was reclaimed: ending it", ua.Port) |
| + orphan.Close() |
| + } else { |
| + s.log("relay: allocation created on port %d with no socket of ours", ua.Port) |
| + } |
| return |
| } |
| delete(s.pending, ua.Port) |
| @@ -610,7 +764,7 @@ func (s *Server) created(src, _ net.Addr, _, _, _ string, relayAddr net.Addr, _ |
| c.b.Incarnation = rec.u.Incarnation |
| c.b.Src = key |
| s.seq++ |
| - e := &allocEntry{conn: c, id: c.b.ID, key: key, seq: s.seq} |
| + e := &allocEntry{conn: c, id: c.b.ID, key: key, seq: s.seq, flow: p.flow} |
| s.allocs[key] = e |
| s.owners[c] = e |
| s.bound[key] = c.b |
| @@ -643,11 +797,17 @@ func (s *Server) release(port int, c *relayConn) { |
| s.ports.remove(port) |
| var id Identity |
| var release bool |
| + var last int64 |
| s.mu.Lock() |
| if p, ok := s.pending[port]; ok && p.conn == c { |
| delete(s.pending, port) |
| + last = p.flow.flush() |
| + } |
| + if cur, ok := s.byPort[port]; ok && cur == c { |
| + delete(s.byPort, port) |
| } |
| if e, ok := s.owners[c]; ok { |
| + last = e.flow.flush() |
| delete(s.owners, c) |
| if e.timer != nil { |
| e.timer.Stop() |
| @@ -664,6 +824,10 @@ func (s *Server) release(port int, c *relayConn) { |
| if release { |
| s.adm.Release(id) |
| } |
| + if last > 0 { |
| + // The last peer packet was read and never forwarded. |
| + s.countRejected(c.b.ID, last) |
| + } |
| } |
| // permit is pion's PermissionHandler: it sees only a peer IP, so the port |
| @@ -692,6 +856,13 @@ func (s *Server) permit(client net.Addr, peer net.IP) bool { |
| if s.cfg.RelayIP != nil && peer.Equal(s.cfg.RelayIP) { |
| return true |
| } |
| + // ExtraAllow is the operator's (and the test fixture's) way to permit |
| + // a range the built-in list denies. Empty on a real box. |
| + for _, n := range s.allow { |
| + if n.Contains(peer) { |
| + return true |
| + } |
| + } |
| for _, n := range s.deny { |
| if n.Contains(peer) { |
| return deny("denied range " + n.String()) |
| @@ -729,13 +900,20 @@ func (s *Server) countOut(id Identity, n int) { |
| } |
| } |
| -// countRejected is what we dropped rather than forwarded: a packet aimed at |
| -// our own EIP on a port that is not a live allocation. Never a budget. |
| -func (s *Server) countRejected(id Identity, n int) { |
| +// countRejected is spec §3's `rejected`: peer bytes that reached a relay |
| +// socket and were dropped for want of a permission. Metered, never charged |
| +// to a budget. |
| +func (s *Server) countRejected(id Identity, n int64) { |
| if s.cfg.Meter != nil { |
| - s.cfg.Meter.Add(id, 0, 0, int64(n)) |
| + s.cfg.Meter.Add(id, 0, 0, n) |
| } |
| - s.countRefusal(id, "eip port") |
| +} |
| + |
| +// countGated is the other kind of drop: a client packet aimed at our own |
| +// EIP on a port that is not a live allocation. It is a policy refusal, not |
| +// a meter line, so it is counted per app and goes no further. |
| +func (s *Server) countGated(id Identity, _ int) { |
| + s.countRefusal(id, "gated") |
| } |
| // overBudget ends an app's allocations once its rolling-hour budget is |
| @@ -757,9 +935,29 @@ func (s *Server) overBudget(id Identity) { |
| }() |
| } |
| +// attrib is which identity a refusal may be counted against. pion hands |
| +// the AuthHandler whatever username arrived, before any integrity check, |
| +// so an unauthenticated sender picks the account and app strings. Only an |
| +// app the authority record actually names gets a row of its own; every |
| +// other refusal shares the zero identity. |
| +func (s *Server) attrib(id Identity) Identity { |
| + if id.Account == "" || id.App == "" { |
| + return Identity{} |
| + } |
| + if _, ok := s.cfg.Authority().Lookup(id.Account, id.App); !ok { |
| + return Identity{} |
| + } |
| + return id |
| +} |
| + |
| func (s *Server) countRefusal(id Identity, reason string) { |
| + id = s.attrib(id) |
| s.mu.Lock() |
| m := s.refusals[id] |
| + if m == nil && len(s.refusals) >= maxRefusalIdentities { |
| + id = Identity{} |
| + m = s.refusals[id] |
| + } |
| if m == nil { |
| m = map[string]int{} |
| s.refusals[id] = m |
| @@ -768,11 +966,31 @@ func (s *Server) countRefusal(id Identity, reason string) { |
| s.mu.Unlock() |
| } |
| +// mayLog rate-limits one identity's one refusal reason to a line a minute, |
| +// so a flood of refusals cannot be turned into a flood of log. |
| +func (s *Server) mayLog(id Identity, reason string) bool { |
| + k := logKey{id: s.attrib(id), reason: reason} |
| + now := s.now() |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if at, ok := s.logged[k]; ok { |
| + if now.Sub(at) < refusalLogEvery { |
| + return false |
| + } |
| + } else if len(s.logged) >= maxLoggedRefusals { |
| + return false |
| + } |
| + s.logged[k] = now |
| + return true |
| +} |
| + |
| // refused counts a refusal against the app and logs why. Its return values |
| // are the AuthHandler's, so a refusal is one line at the call site. |
| func (s *Server) refused(id Identity, reason, format string, args ...any) ([]byte, bool) { |
| s.countRefusal(id, reason) |
| - s.log(format, args...) |
| + if s.mayLog(id, reason) { |
| + s.log(format, args...) |
| + } |
| return nil, false |
| } |
| @@ -782,6 +1000,12 @@ func (s *Server) refusalCount(id Identity, reason string) int { |
| return s.refusals[id][reason] |
| } |
| +func (s *Server) refusalIdentities() int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return len(s.refusals) |
| +} |
| + |
| // entryFor is the owner binding lookup the out leg is attributed through. |
| func (s *Server) entryFor(key string) *allocEntry { |
| s.mu.Lock() |
| @@ -801,23 +1025,56 @@ type listenerConn struct { |
| } |
| func (c *listenerConn) WriteTo(p []byte, addr net.Addr) (int, error) { |
| + // A TURN response is the client asking us something, not relayed |
| + // media: bucketing it would drop Refresh responses at the app's rate |
| + // cap and end calls for a reason spec §3 does not list, and metering |
| + // it would bill the client for its own signalling. |
| + if !relayedPayload(p) { |
| + return c.PacketConn.WriteTo(p, addr) |
| + } |
| e := c.s.entryFor(typedSrc(addr)) |
| if e != nil { |
| // Peek both, then spend both: a box refusal must not leave the |
| // app's bucket debited (see relayConn.WriteTo). |
| - if !e.conn.appBucket.Peek(len(p)) || !c.s.box.Peek(len(p)) { |
| + if e.conn.appBucket != nil && !e.conn.appBucket.Peek(len(p)) { |
| return len(p), nil |
| } |
| - e.conn.appBucket.Spend(len(p)) |
| - c.s.box.Spend(len(p)) |
| + if c.s.box != nil && !c.s.box.Peek(len(p)) { |
| + return len(p), nil |
| + } |
| + if e.conn.appBucket != nil { |
| + e.conn.appBucket.Spend(len(p)) |
| + } |
| + if c.s.box != nil { |
| + c.s.box.Spend(len(p)) |
| + } |
| } |
| n, err := c.PacketConn.WriteTo(p, addr) |
| if n > 0 && e != nil { |
| + e.flow.forwarded() |
| c.s.countOut(e.id, n) |
| } |
| return n, err |
| } |
| +// relayedPayload reports whether a packet bound for a client carries peer |
| +// media rather than TURN signalling: a ChannelData message (first two bits |
| +// 0b01, so 0x40 to 0x7f) or a Data indication. |
| +func relayedPayload(p []byte) bool { |
| + if len(p) == 0 { |
| + return false |
| + } |
| + if p[0]&0xc0 == 0x40 { |
| + return true |
| + } |
| + if !stun.IsMessage(p) { |
| + return false |
| + } |
| + var t stun.MessageType |
| + t.ReadValue(binary.BigEndian.Uint16(p[0:2])) |
| + return t.Method == stun.MethodData && t.Class == stun.ClassIndication |
| +} |
| + |
| // countingListener wraps the TCP listener so the out leg is metered for |
| // TCP clients too. A TCP stream cannot have bytes dropped out of it the |
| // way a datagram can, so the buckets are spent but never enforced here. |
| @@ -841,11 +1098,17 @@ type countingConn struct { |
| } |
| func (c *countingConn) Write(p []byte) (int, error) { |
| + payload := relayedPayload(p) |
| n, err := c.Conn.Write(p) |
| - if n > 0 { |
| + if n > 0 && payload { |
| if e := c.s.entryFor(c.key); e != nil { |
| - e.conn.appBucket.Spend(n) |
| - c.s.box.Spend(n) |
| + if e.conn.appBucket != nil { |
| + e.conn.appBucket.Spend(n) |
| + } |
| + if c.s.box != nil { |
| + c.s.box.Spend(n) |
| + } |
| + e.flow.forwarded() |
| c.s.countOut(e.id, n) |
| } |
| } |
diff --git a/internal/relay/server_test.go b/internal/relay/server_test.go| index 375245c4e..3ba4b66f6 100644 |
| --- a/internal/relay/server_test.go |
| +++ b/internal/relay/server_test.go |
| @@ -28,7 +28,8 @@ func testMaster() []byte { |
| // loopback sockets must advertise something else, because the EIP port |
| // gate (spec §3) drops anything aimed at the advertised address on a port |
| // that is not a live allocation — and on loopback every peer is on it. |
| -func newTestServer(t *testing.T, relayIP string, meter *fakeMeter) (s *Server, setNow func(time.Time), now func() time.Time, masters *Masters, auth *Authority) { |
| +// tweak, when given, is applied to the config before the server starts. |
| +func newTestServer(t *testing.T, relayIP string, meter *fakeMeter, tweak func(*ServerConfig)) (s *Server, setNow func(time.Time), now func() time.Time, masters *Masters, auth *Authority) { |
| t.Helper() |
| master := testMaster() |
| masters, err := ParseMasters(strings.NewReader("k1 " + b64(master) + " -\n")) |
| @@ -66,6 +67,9 @@ func newTestServer(t *testing.T, relayIP string, meter *fakeMeter) (s *Server, s |
| if meter != nil { |
| cfg.Meter = meter |
| } |
| + if tweak != nil { |
| + tweak(&cfg) |
| + } |
| s, err = NewServer(cfg) |
| if err != nil { |
| t.Fatal(err) |
| @@ -128,7 +132,7 @@ func waitFor(t *testing.T, cond func() bool) { |
| } |
| func TestServerEndToEnd(t *testing.T) { |
| - s, setNow, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, setNow, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| @@ -155,7 +159,9 @@ func TestServerEndToEnd(t *testing.T) { |
| if _, err := turnClient(t, s.UDPAddr(), otherUser, otherPass).Allocate(); err == nil { |
| t.Fatal("unknown app allocated") |
| } |
| - if s.refusalCount(Identity{"abc", "docs"}, "unknown app") != 1 { |
| + // It is counted against the shared unknown bucket, not against an |
| + // identity the caller made up. |
| + if s.refusalCount(Identity{}, "unknown app") != 1 { |
| t.Fatalf("refusals = %v", s.refusals) |
| } |
| @@ -169,12 +175,8 @@ func TestServerEndToEnd(t *testing.T) { |
| // and checks both meter legs. It advertises an EIP that is not the peers' |
| // address, as a real box does. |
| func TestServerRelaysBothLegs(t *testing.T) { |
| - allowLoopbackPeers(t) |
| meter := &fakeMeter{} |
| - s, _, now, _, _ := newTestServer(t, "203.0.113.5", meter) |
| - // The fixture listens on loopback, so its own listener address is the |
| - // peers' address too; a real box's listener is not on the peer's net. |
| - s.local = nil |
| + s, _, now, _, _ := newTestServer(t, "203.0.113.5", meter, allowLoopbackPeers) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| relayConn, err := client.Allocate() |
| @@ -225,19 +227,11 @@ func TestServerRelaysBothLegs(t *testing.T) { |
| } |
| } |
| -// allowLoopbackPeers drops loopback from the built-in deny list for one |
| -// test. A relay on a real box never permits a loopback peer; a test that |
| -// carries traffic between two sockets on this machine has nowhere else to |
| -// put them, which is why builtinDeny is a var. |
| -func allowLoopbackPeers(t *testing.T) { |
| - t.Helper() |
| - saved := builtinDeny |
| - builtinDeny = mustCIDRs( |
| - "169.254.0.0/16", "fe80::/10", |
| - "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", |
| - "fc00::/7", "100.64.0.0/10", |
| - ) |
| - t.Cleanup(func() { builtinDeny = saved }) |
| +// allowLoopbackPeers is the ExtraAllow seam: a relay on a real box never |
| +// permits a loopback peer, and a test that carries traffic between two |
| +// sockets on this machine has nowhere else to put them. |
| +func allowLoopbackPeers(cfg *ServerConfig) { |
| + cfg.ExtraAllow = mustCIDRs("127.0.0.0/8") |
| } |
| // fakeMeter is the Task 7 meter's shape, which ServerConfig.Meter names as |
| @@ -265,7 +259,7 @@ func (m *fakeMeter) totals(id Identity) (in, out, rejected int64) { |
| } |
| func TestServerRefusesPrivatePeer(t *testing.T) { |
| - s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| if _, err := client.Allocate(); err != nil { |
| @@ -277,7 +271,7 @@ func TestServerRefusesPrivatePeer(t *testing.T) { |
| } |
| func TestServerEndWhere(t *testing.T) { |
| - s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| if _, err := client.Allocate(); err != nil { |
| @@ -291,7 +285,7 @@ func TestServerEndWhere(t *testing.T) { |
| } |
| func TestServerShedOldest(t *testing.T) { |
| - s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| first := turnClient(t, s.UDPAddr(), username, password) |
| if _, err := first.Allocate(); err != nil { |
| @@ -312,7 +306,7 @@ func TestServerShedOldest(t *testing.T) { |
| // only refuses new allocations bounds nothing: seeded past its hourly |
| // budget, the app's live allocation goes too. |
| func TestServerBudgetEndsAllocations(t *testing.T) { |
| - s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| if _, err := client.Allocate(); err != nil { |
| @@ -324,12 +318,9 @@ func TestServerBudgetEndsAllocations(t *testing.T) { |
| } |
| func TestServerAuthRefusals(t *testing.T) { |
| - s, _, now, masters, auth := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, masters, auth := newTestServer(t, "127.0.0.1", nil, nil) |
| src := &net.UDPAddr{IP: net.ParseIP("192.0.2.7"), Port: 5000} |
| - sign := func(username, account, app, incarnation string) string { |
| - return Credential(DeriveKey(testMaster(), account, app, incarnation), username) |
| - } |
| ok := func(username string) bool { |
| key, ok := s.auth(username, "carlos", src) |
| return ok && len(key) > 0 |
| @@ -352,7 +343,7 @@ func TestServerAuthRefusals(t *testing.T) { |
| if ok(c.username) { |
| t.Fatalf("%s: accepted", c.name) |
| } |
| - if s.refusalCount(refusalIdentity(c.username), c.reason) == 0 { |
| + if s.refusalCount(s.attrib(refusalIdentity(c.username)), c.reason) == 0 { |
| t.Fatalf("%s: reason %q not counted (%v)", c.name, c.reason, s.refusals) |
| } |
| } |
| @@ -381,7 +372,6 @@ func TestServerAuthRefusals(t *testing.T) { |
| s.mu.Unlock() |
| auth.Apps = append(auth.Apps, AuthorityApp{Account: "abc", App: "docs", Incarnation: "0123456789abcdef", Regions: []string{"test"}}) |
| docs := fmt.Sprintf("%d:%s", now().Add(time.Hour).Unix(), Suffix("abc", "docs", "0123456789abcdef", "k1")) |
| - _ = sign(docs, "abc", "docs", "0123456789abcdef") |
| if ok(docs) { |
| t.Fatal("a second app allocated from a bound source") |
| } |
| @@ -399,6 +389,177 @@ func TestServerAuthRefusals(t *testing.T) { |
| } |
| } |
| +// TestServerRefusalStateIsBounded is the memory-exhaustion case: pion |
| +// hands the AuthHandler the raw username before it checks |
| +// MESSAGE-INTEGRITY, and the nonce is not source-bound, so an unauthenticated |
| +// sender chooses the account and app strings. None of that may become a |
| +// map entry of its own. |
| +func TestServerRefusalStateIsBounded(t *testing.T) { |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| + src := &net.UDPAddr{IP: net.ParseIP("192.0.2.10"), Port: 6000} |
| + exp := now().Add(time.Hour).Unix() |
| + const flood = 10000 |
| + for i := 0; i < flood; i++ { |
| + username := fmt.Sprintf("%d:%s", exp, Suffix(fmt.Sprintf("a%d", i), "meet", "0123456789abcdef", "k1")) |
| + if _, ok := s.auth(username, "carlos", src); ok { |
| + t.Fatalf("made-up account %d authenticated", i) |
| + } |
| + } |
| + if n := s.refusalIdentities(); n > 2 { |
| + t.Fatalf("refusal identities = %d, want at most 2", n) |
| + } |
| + if got := s.refusalCount(Identity{}, "unknown app"); got != flood { |
| + t.Fatalf("unknown-app refusals = %d, want %d", got, flood) |
| + } |
| + s.mu.Lock() |
| + auths, logs := len(s.lastAuth), len(s.logged) |
| + s.mu.Unlock() |
| + if auths != 0 { |
| + t.Fatalf("lastAuth = %d, want 0: nothing authenticated", auths) |
| + } |
| + if logs > 2 { |
| + t.Fatalf("log records = %d, want the refusal log rate-limited", logs) |
| + } |
| +} |
| + |
| +// TestServerAuthMemoryIsCapped covers the other half: usernames that do |
| +// verify are remembered for the allocation event, but only so many. |
| +func TestServerAuthMemoryIsCapped(t *testing.T) { |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| + username, _ := creds(now(), "abc", "meet", "0123456789abcdef") |
| + for i := 0; i < s.maxLastAuth+600; i++ { |
| + src := &net.UDPAddr{IP: net.ParseIP("192.0.2.11"), Port: 1024 + i} |
| + if _, ok := s.auth(username, "carlos", src); !ok { |
| + t.Fatalf("source %d refused", i) |
| + } |
| + } |
| + s.mu.Lock() |
| + n := len(s.lastAuth) |
| + s.mu.Unlock() |
| + if n > s.maxLastAuth { |
| + t.Fatalf("lastAuth = %d, want at most %d", n, s.maxLastAuth) |
| + } |
| +} |
| + |
| +// TestServerCountsRejectedPeerBytes is spec §3's `rejected`: peer bytes |
| +// that reached a relay socket and were dropped for want of a permission. |
| +func TestServerCountsRejectedPeerBytes(t *testing.T) { |
| + meter := &fakeMeter{} |
| + s, _, now, _, _ := newTestServer(t, "203.0.113.5", meter, nil) |
| + username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| + client := turnClient(t, s.UDPAddr(), username, password) |
| + relayed, err := client.Allocate() |
| + if err != nil { |
| + t.Fatalf("allocate: %v", err) |
| + } |
| + defer relayed.Close() |
| + port := relayed.LocalAddr().(*net.UDPAddr).Port |
| + |
| + peer, err := net.ListenPacket("udp4", "127.0.0.1:0") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer peer.Close() |
| + // No permission was ever created for this peer, so pion drops both. |
| + to := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port} |
| + for i := 0; i < 2; i++ { |
| + if _, err := peer.WriteTo(make([]byte, 100), to); err != nil { |
| + t.Fatalf("peer write: %v", err) |
| + } |
| + } |
| + id := Identity{"abc", "meet"} |
| + waitFor(t, func() bool { _, _, rejected := meter.totals(id); return rejected >= 100 }) |
| + |
| + s.EndAll("test") |
| + waitFor(t, func() bool { return s.Live() == 0 }) |
| + in, out, rejected := meter.totals(id) |
| + if rejected != 200 { |
| + t.Fatalf("rejected = %d, want 200", rejected) |
| + } |
| + if out != 0 { |
| + t.Fatalf("out = %d, want 0: nothing was forwarded to the client", out) |
| + } |
| + if in != 0 { |
| + t.Fatalf("in = %d, want 0", in) |
| + } |
| +} |
| + |
| +// TestServerControlIsNotBucketed: at a saturated per-allocation bucket a |
| +// client must still be able to allocate and refresh. Only relayed payload |
| +// is bucketed and metered. |
| +func TestServerControlIsNotBucketed(t *testing.T) { |
| + meter := &fakeMeter{} |
| + s, _, now, _, _ := newTestServer(t, "203.0.113.5", meter, func(cfg *ServerConfig) { |
| + cfg.Limits.AllocBitsPerSec = 8 // one byte a second: every control response would be dropped |
| + cfg.Limits.BoxBitsPerSec = 8 |
| + }) |
| + username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| + client := turnClient(t, s.UDPAddr(), username, password) |
| + relayed, err := client.Allocate() |
| + if err != nil { |
| + t.Fatalf("allocate at a saturated bucket: %v", err) |
| + } |
| + defer relayed.Close() |
| + // A second round trip on the same listener, after the allocation |
| + // exists: its response must get through too. |
| + if _, err := client.SendBindingRequest(); err != nil { |
| + t.Fatalf("binding request at a saturated bucket: %v", err) |
| + } |
| + if in, out, rejected := meter.totals(Identity{"abc", "meet"}); in|out|rejected != 0 { |
| + t.Fatalf("signalling was metered: in=%d out=%d rejected=%d", in, out, rejected) |
| + } |
| +} |
| + |
| +// TestServerClosesOrphanedAllocation: if the pending socket is gone by the |
| +// time pion reports the allocation, pion owns an allocation no deadline of |
| +// ours can end. It must be ended on the spot. |
| +func TestServerClosesOrphanedAllocation(t *testing.T) { |
| + s, setNow, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| + src := &net.UDPAddr{IP: net.ParseIP("192.0.2.12"), Port: 7000} |
| + |
| + // The socket is still ours, but its pending record has gone. |
| + _, addr, err := s.AllocatePacketConn("udp4", 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + port := addr.(*net.UDPAddr).Port |
| + username, _ := creds(now(), "abc", "meet", "0123456789abcdef") |
| + if _, ok := s.auth(username, "carlos", src); !ok { |
| + t.Fatal("fixture credential refused") |
| + } |
| + s.mu.Lock() |
| + delete(s.pending, port) |
| + s.mu.Unlock() |
| + s.created(src, nil, "udp", username, "carlos", addr, 0) |
| + if s.Live() != 0 { |
| + t.Fatalf("live = %d, want the orphan ended", s.Live()) |
| + } |
| + if s.ports.has(port) { |
| + t.Fatal("the orphaned socket was left open") |
| + } |
| + |
| + // And the real race: the sweep reclaimed the socket before pion |
| + // reported the allocation. |
| + _, addr2, err := s.AllocatePacketConn("udp4", 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + port2 := addr2.(*net.UDPAddr).Port |
| + setNow(now().Add(pendingLife + time.Second)) |
| + s.tick() |
| + if s.ports.has(port2) { |
| + t.Fatal("the sweep left the socket open") |
| + } |
| + username2, _ := creds(now(), "abc", "meet", "0123456789abcdef") |
| + if _, ok := s.auth(username2, "carlos", src); !ok { |
| + t.Fatal("fixture credential refused after the clock moved") |
| + } |
| + s.created(src, nil, "udp", username2, "carlos", addr2, 0) |
| + if s.Live() != 0 { |
| + t.Fatalf("live = %d, want no allocation for a reclaimed socket", s.Live()) |
| + } |
| +} |
| + |
| // refusalIdentity is the identity a refusal is attributed to: the parsed |
| // username's, or the zero identity when the username would not parse. |
| func refusalIdentity(username string) Identity { |
| @@ -450,7 +611,7 @@ func TestServerPermit(t *testing.T) { |
| // a replacement allocation on the same typed source can exist while the old |
| // wrapper is still closing. The old wrapper's close must not erase it. |
| func TestServerCloseKeepsReplacementBinding(t *testing.T) { |
| - s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil) |
| + s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, nil) |
| username, password := creds(now(), "abc", "meet", "0123456789abcdef") |
| client := turnClient(t, s.UDPAddr(), username, password) |
| if _, err := client.Allocate(); err != nil { |