| new file mode 100644 |
| index 000000000..403d01380 |
| --- /dev/null |
| +++ b/internal/relay/server.go |
| @@ -0,0 +1,862 @@ |
| +package relay |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "math/rand/v2" |
| + "net" |
| + "sort" |
| + "strconv" |
| + "sync" |
| + "time" |
| + |
| + "github.com/pion/logging" |
| + "github.com/pion/turn/v4" |
| +) |
| + |
| +// Realm is the TURN realm every credential is minted and verified under. |
| +const Realm = "carlos" |
| + |
| +// pendingLife is how long a socket or an admission may sit unclaimed |
| +// before tick reclaims it. Both are consumed within one request in the |
| +// normal path; this only catches an allocation that failed after pion had |
| +// already asked us for a socket. |
| +const pendingLife = 30 * time.Second |
| + |
| +// authMemory is how long the last parsed username for a typed source is |
| +// kept once that source has no allocation, so a client that authenticates |
| +// and never allocates cannot grow the map without bound. |
| +const authMemory = 10 * time.Minute |
| + |
| +var errTCPAllocUnsupported = errors.New("relay: TCP relay allocations are not supported") |
| + |
| +// ServerConfig is everything the relay needs from its process: where to |
| +// listen, what to advertise, and the four live views (masters, authority, |
| +// lease, limits) it reads on every request. |
| +type ServerConfig struct { |
| + Region string |
| + RelayIP net.IP // the EIP advertised to clients |
| + 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 |
| + Masters func() *Masters |
| + Authority func() *Authority |
| + Lease func(now time.Time) bool |
| + Limits Limits |
| + Now func() time.Time |
| + Log func(string, ...any) |
| + |
| + // Meter is the byte meter (Task 7); nil disables metering. It is an |
| + // interface so this file does not depend on the meter's shape. |
| + Meter interface { |
| + Add(id Identity, in, out, rejected int64) |
| + } |
| +} |
| + |
| +// Server is the relay: a pion TURN server whose every socket we own, so |
| +// that an allocation can be ended, attributed and bounded from outside |
| +// pion, which exposes no way to do any of it. |
| +type Server struct { |
| + cfg ServerConfig |
| + 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 |
| + |
| + turn *turn.Server |
| + udp net.PacketConn |
| + tcp net.Listener |
| + listeners []string |
| + |
| + mu sync.Mutex |
| + seq int64 |
| + pending map[int]pendingConn // relay port -> socket awaiting OnAllocationCreated |
| + lastAuth map[string]authRec // typed source -> the username it last authenticated with |
| + admits map[string]admitRec // typed source -> an admission not yet owned by an allocation |
| + 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 |
| + refusals map[Identity]map[string]int |
| + shedding map[Identity]bool |
| +} |
| + |
| +type allocEntry struct { |
| + conn *relayConn |
| + id Identity |
| + key string // the typed source this allocation was created for |
| + seq int64 |
| + timer *time.Timer |
| +} |
| + |
| +type pendingConn struct { |
| + conn *relayConn |
| + at time.Time |
| +} |
| + |
| +type authRec struct { |
| + u Username |
| + at time.Time |
| +} |
| + |
| +type admitRec struct { |
| + id Identity |
| + at time.Time |
| +} |
| + |
| +// builtinDeny is the peer policy every relay carries whatever the operator |
| +// configures: loopback, link-local (the IMDS address included), RFC 1918, |
| +// unique-local and carrier-grade NAT space. |
| +var builtinDeny = mustCIDRs( |
| + "127.0.0.0/8", "::1/128", |
| + "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", |
| +) |
| + |
| +func mustCIDRs(ss ...string) []*net.IPNet { |
| + out := make([]*net.IPNet, 0, len(ss)) |
| + for _, s := range ss { |
| + _, n, err := net.ParseCIDR(s) |
| + if err != nil { |
| + panic("relay: bad built-in CIDR " + s + ": " + err.Error()) |
| + } |
| + out = append(out, n) |
| + } |
| + return out |
| +} |
| + |
| +func NewServer(cfg ServerConfig) (*Server, error) { |
| + if cfg.Now == nil { |
| + cfg.Now = time.Now |
| + } |
| + if cfg.Log == nil { |
| + cfg.Log = func(string, ...any) {} |
| + } |
| + if cfg.MinPort == 0 { |
| + cfg.MinPort = 49152 |
| + } |
| + if cfg.MaxPort == 0 { |
| + cfg.MaxPort = 65535 |
| + } |
| + switch { |
| + case cfg.RelayIP == nil: |
| + return nil, errors.New("relay: RelayIP is required") |
| + case cfg.MinPort > cfg.MaxPort: |
| + return nil, fmt.Errorf("relay: port range %d-%d is empty", cfg.MinPort, cfg.MaxPort) |
| + case cfg.Masters == nil || cfg.Authority == nil || cfg.Lease == nil: |
| + return nil, errors.New("relay: Masters, Authority and Lease are required") |
| + case cfg.UDP == "" && cfg.TCP == "": |
| + return nil, errors.New("relay: no listener configured") |
| + } |
| + |
| + 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{}, |
| + } |
| + |
| + var packets []turn.PacketConnConfig |
| + var listeners []turn.ListenerConfig |
| + if cfg.UDP != "" { |
| + pc, err := net.ListenPacket("udp", cfg.UDP) |
| + if err != nil { |
| + return nil, fmt.Errorf("relay: listen udp %s: %w", cfg.UDP, err) |
| + } |
| + s.udp = pc |
| + s.noteLocal(pc.LocalAddr()) |
| + s.listeners = append(s.listeners, "udp:"+portOf(pc.LocalAddr())) |
| + packets = append(packets, turn.PacketConnConfig{ |
| + PacketConn: &listenerConn{PacketConn: pc, s: s}, |
| + RelayAddressGenerator: s, |
| + PermissionHandler: s.permit, |
| + }) |
| + } |
| + if cfg.TCP != "" { |
| + l, err := net.Listen("tcp", cfg.TCP) |
| + if err != nil { |
| + if s.udp != nil { |
| + s.udp.Close() |
| + } |
| + return nil, fmt.Errorf("relay: listen tcp %s: %w", cfg.TCP, err) |
| + } |
| + s.tcp = l |
| + s.noteLocal(l.Addr()) |
| + s.listeners = append(s.listeners, "tcp:"+portOf(l.Addr())) |
| + listeners = append(listeners, turn.ListenerConfig{ |
| + Listener: &countingListener{Listener: l, s: s}, |
| + RelayAddressGenerator: s, |
| + PermissionHandler: s.permit, |
| + }) |
| + } |
| + |
| + ts, err := turn.NewServer(turn.ServerConfig{ |
| + Realm: Realm, |
| + AuthHandler: s.auth, |
| + QuotaHandler: s.quota, |
| + EventHandler: turn.EventHandler{OnAllocationCreated: s.created}, |
| + PacketConnConfigs: packets, |
| + ListenerConfigs: listeners, |
| + LoggerFactory: logging.NewDefaultLoggerFactory(), |
| + }) |
| + if err != nil { |
| + if s.udp != nil { |
| + s.udp.Close() |
| + } |
| + if s.tcp != nil { |
| + s.tcp.Close() |
| + } |
| + return nil, fmt.Errorf("relay: pion server: %w", err) |
| + } |
| + s.turn = ts |
| + return s, nil |
| +} |
| + |
| +func (s *Server) noteLocal(a net.Addr) { |
| + var ip net.IP |
| + switch v := a.(type) { |
| + case *net.UDPAddr: |
| + ip = v.IP |
| + case *net.TCPAddr: |
| + ip = v.IP |
| + } |
| + if ip != nil && !ip.IsUnspecified() { |
| + s.local = append(s.local, ip) |
| + } |
| +} |
| + |
| +func portOf(a net.Addr) string { |
| + switch v := a.(type) { |
| + case *net.UDPAddr: |
| + return strconv.Itoa(v.Port) |
| + case *net.TCPAddr: |
| + return strconv.Itoa(v.Port) |
| + } |
| + _, p, err := net.SplitHostPort(a.String()) |
| + if err != nil { |
| + return a.String() |
| + } |
| + return p |
| +} |
| + |
| +// Listeners names the transports that are up, e.g. ["udp:3478","tcp:3478"], |
| +// for the heartbeat. |
| +func (s *Server) Listeners() []string { return append([]string{}, s.listeners...) } |
| + |
| +// UDPAddr is the UDP listener's bound address, or nil if UDP is disabled. |
| +func (s *Server) UDPAddr() net.Addr { |
| + if s.udp == nil { |
| + return nil |
| + } |
| + return s.udp.LocalAddr() |
| +} |
| + |
| +// Live is the number of allocations the relay is carrying. |
| +func (s *Server) Live() int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return len(s.allocs) |
| +} |
| + |
| +// Seed replays metered bytes into the rolling-hour windows, so a restart |
| +// resumes its budgets from the metrics objects it wrote. |
| +func (s *Server) Seed(id Identity, at time.Time, n int64) { s.adm.Seed(id, at, n) } |
| + |
| +// Close ends every allocation and stops listening. |
| +func (s *Server) Close() error { |
| + s.EndAll("server closing") |
| + s.mu.Lock() |
| + pending := make([]*relayConn, 0, len(s.pending)) |
| + for _, p := range s.pending { |
| + pending = append(pending, p.conn) |
| + } |
| + s.mu.Unlock() |
| + for _, c := range pending { |
| + c.Close() |
| + } |
| + if s.turn == nil { |
| + return nil |
| + } |
| + return s.turn.Close() |
| +} |
| + |
| +// EndAll ends every allocation: a lease lapse, or shutdown. |
| +func (s *Server) EndAll(reason string) { |
| + s.EndWhere(func(*Binding) bool { return true }, reason) |
| +} |
| + |
| +// EndWhere ends every allocation whose binding matches: a disable, an |
| +// incarnation change, this region leaving an app's list, a byte budget, a |
| +// retired master. The predicate is called on a copy of the binding, off |
| +// the server's lock. |
| +func (s *Server) EndWhere(match func(*Binding) bool, reason string) { |
| + type candidate struct { |
| + conn *relayConn |
| + b Binding |
| + } |
| + s.mu.Lock() |
| + cands := make([]candidate, 0, len(s.allocs)) |
| + for _, e := range s.allocs { |
| + cands = append(cands, candidate{conn: e.conn, b: *e.conn.b}) |
| + } |
| + s.mu.Unlock() |
| + |
| + n := 0 |
| + for _, c := range cands { |
| + b := c.b |
| + if match(&b) { |
| + c.conn.Close() |
| + n++ |
| + } |
| + } |
| + if n > 0 { |
| + s.log("relay: ended %d allocation(s): %s", n, reason) |
| + } |
| +} |
| + |
| +// ShedOldest ends the n oldest allocations, for memory pressure. |
| +func (s *Server) ShedOldest(n int) { |
| + if n <= 0 { |
| + return |
| + } |
| + s.mu.Lock() |
| + entries := make([]*allocEntry, 0, len(s.allocs)) |
| + for _, e := range s.allocs { |
| + entries = append(entries, e) |
| + } |
| + s.mu.Unlock() |
| + sort.Slice(entries, func(i, j int) bool { return entries[i].seq < entries[j].seq }) |
| + if n > len(entries) { |
| + n = len(entries) |
| + } |
| + for _, e := range entries[:n] { |
| + e.conn.Close() |
| + } |
| + if n > 0 { |
| + s.log("relay: shed %d oldest allocation(s)", n) |
| + } |
| +} |
| + |
| +// tick is the sweep: it ends allocations whose credential has expired |
| +// (belt for the per-allocation timers), whose master has retired, whose app |
| +// or account is over budget, and every allocation at all if the serving |
| +// lease has lapsed. It also reclaims sockets and admissions that a failed |
| +// allocation left behind. |
| +func (s *Server) tick() { |
| + now := s.now() |
| + masters := s.cfg.Masters() |
| + leaseDead := !s.cfg.Lease(now) |
| + |
| + type victim struct { |
| + conn *relayConn |
| + reason string |
| + } |
| + var victims []victim |
| + var stale []*relayConn |
| + |
| + // A masters file that failed to reload leaves masters nil; that is a |
| + // reason to keep serving on the keys we have, not to end anything. |
| + retired := func(kid string) bool { |
| + if masters == nil || kid == "" { |
| + return false |
| + } |
| + _, ok := masters.Lookup(kid, now) |
| + return !ok |
| + } |
| + |
| + s.mu.Lock() |
| + for _, e := range s.allocs { |
| + b := e.conn.b |
| + switch { |
| + case leaseDead: |
| + victims = append(victims, victim{e.conn, "serving lease lapsed"}) |
| + case !b.Expiry.IsZero() && !b.Expiry.After(now): |
| + victims = append(victims, victim{e.conn, "credential expired"}) |
| + case retired(b.Kid): |
| + victims = append(victims, victim{e.conn, "master " + b.Kid + " retired"}) |
| + case s.adm.OverBudget(e.id, now): |
| + victims = append(victims, victim{e.conn, "byte budget"}) |
| + } |
| + } |
| + for port, p := range s.pending { |
| + if now.Sub(p.at) > pendingLife { |
| + delete(s.pending, port) |
| + stale = append(stale, p.conn) |
| + } |
| + } |
| + var lapsed []Identity |
| + for key, a := range s.admits { |
| + if now.Sub(a.at) > pendingLife { |
| + delete(s.admits, key) |
| + lapsed = append(lapsed, a.id) |
| + } |
| + } |
| + for key, a := range s.lastAuth { |
| + if _, live := s.allocs[key]; !live && now.Sub(a.at) > authMemory { |
| + delete(s.lastAuth, key) |
| + } |
| + } |
| + s.mu.Unlock() |
| + |
| + for _, id := range lapsed { |
| + s.adm.Release(id) |
| + } |
| + for _, c := range stale { |
| + c.Close() |
| + } |
| + for _, v := range victims { |
| + s.log("relay: ending allocation for %s/%s: %s", v.conn.b.ID.Account, v.conn.b.ID.App, v.reason) |
| + v.conn.Close() |
| + } |
| +} |
| + |
| +// --- turn.RelayAddressGenerator ------------------------------------------ |
| + |
| +func (s *Server) Validate() error { |
| + if s.cfg.RelayIP == nil { |
| + return errors.New("relay: RelayIP is required") |
| + } |
| + if s.cfg.MinPort > s.cfg.MaxPort { |
| + return fmt.Errorf("relay: port range %d-%d is empty", s.cfg.MinPort, s.cfg.MaxPort) |
| + } |
| + return nil |
| +} |
| + |
| +// AllocatePacketConn binds one allocation's relay socket in the configured |
| +// port range and hands pion our wrapper for it: the wrapper is how the |
| +// allocation is ended, metered and bounded. The advertised address is the |
| +// EIP, which is not an address on the box, so the socket itself binds the |
| +// wildcard. |
| +func (s *Server) AllocatePacketConn(network string, requestedPort int) (net.PacketConn, net.Addr, error) { |
| + if network == "" { |
| + network = "udp4" |
| + } |
| + lo, hi := int(s.cfg.MinPort), int(s.cfg.MaxPort) |
| + span := hi - lo + 1 |
| + tries := span |
| + if tries > 64 { |
| + tries = 64 |
| + } |
| + ports := make([]int, 0, tries+1) |
| + if requestedPort >= lo && requestedPort <= hi { |
| + ports = append(ports, requestedPort) |
| + } |
| + start := rand.IntN(span) |
| + for i := 0; i < tries; i++ { |
| + ports = append(ports, lo+(start+i)%span) |
| + } |
| + |
| + for _, port := range ports { |
| + if s.ports.has(port) { |
| + continue |
| + } |
| + pc, err := net.ListenPacket(network, net.JoinHostPort("", strconv.Itoa(port))) |
| + if err != nil { |
| + 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) }) |
| + s.ports.add(port) |
| + s.mu.Lock() |
| + s.pending[port] = pendingConn{conn: c, at: s.now()} |
| + s.mu.Unlock() |
| + return c, &net.UDPAddr{IP: s.cfg.RelayIP, Port: port}, nil |
| + } |
| + return nil, nil, fmt.Errorf("relay: no free port in %d-%d", lo, hi) |
| +} |
| + |
| +// AllocateConn would be a TCP relay allocation (RFC 6062), which we do not |
| +// offer: a TURN client may reach us over TCP, but every relayed leg is UDP. |
| +func (s *Server) AllocateConn(string, int) (net.Conn, net.Addr, error) { |
| + return nil, nil, errTCPAllocUnsupported |
| +} |
| + |
| +// --- pion handlers ------------------------------------------------------- |
| + |
| +// auth is pion's AuthHandler: called on every authenticated request, and |
| +// the whole of spec §4's verification. It returns the key pion checks |
| +// MESSAGE-INTEGRITY against, so a wrong credential fails inside pion and |
| +// never reaches here. |
| +func (s *Server) auth(username, realm string, src net.Addr) ([]byte, bool) { |
| + now := s.now() |
| + u, err := ParseUsername(username) |
| + if err != nil { |
| + return s.refused(Identity{}, "bad shape", "relay: auth refused from %s: %v", src, err) |
| + } |
| + id := Identity{Account: u.Account, App: u.App} |
| + |
| + if !u.Expiry.After(now) || u.Expiry.After(now.Add(MaxCredentialLife)) { |
| + return s.refused(id, "expired", "relay: auth refused for %s/%s: expiry %s outside (now, now+%s]", |
| + id.Account, id.App, u.Expiry.UTC().Format(time.RFC3339), MaxCredentialLife) |
| + } |
| + |
| + s.mu.Lock() |
| + bound, isBound := s.bound[typedSrc(src)] |
| + s.mu.Unlock() |
| + if isBound && (bound.ID != id || bound.Incarnation != u.Incarnation) { |
| + return s.refused(id, "bound elsewhere", "relay: auth refused for %s/%s: source %s is bound to %s/%s", |
| + id.Account, id.App, typedSrc(src), bound.ID.Account, bound.ID.App) |
| + } |
| + |
| + app, ok := s.cfg.Authority().Lookup(u.Account, u.App) |
| + switch { |
| + case !ok: |
| + return s.refused(id, "unknown app", "relay: auth refused for %s/%s: not in the authority record", id.Account, id.App) |
| + case app.Incarnation != u.Incarnation: |
| + return s.refused(id, "wrong incarnation", "relay: auth refused for %s/%s: incarnation %s, record has %s", |
| + id.Account, id.App, u.Incarnation, app.Incarnation) |
| + case !contains(app.Regions, s.cfg.Region): |
| + return s.refused(id, "wrong region", "relay: auth refused for %s/%s: region %s not in %v", |
| + id.Account, id.App, s.cfg.Region, app.Regions) |
| + } |
| + |
| + masters := s.cfg.Masters() |
| + if masters == nil { |
| + return s.refused(id, "unknown kid", "relay: auth refused for %s/%s: no masters loaded", id.Account, id.App) |
| + } |
| + master, ok := masters.Lookup(u.Kid, now) |
| + if !ok { |
| + return s.refused(id, "unknown kid", "relay: auth refused for %s/%s: kid %s is unknown or retired", |
| + id.Account, id.App, u.Kid) |
| + } |
| + |
| + 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() |
| + return turn.GenerateAuthKey(username, realm, Credential(key, username)), true |
| +} |
| + |
| +// 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. |
| +func (s *Server) quota(username, _ string, src net.Addr) bool { |
| + now := s.now() |
| + if !s.cfg.Lease(now) { |
| + s.log("relay: allocation refused: serving lease has lapsed") |
| + return false |
| + } |
| + u, err := ParseUsername(username) |
| + if err != nil { |
| + return false // auth would already have refused this |
| + } |
| + id := Identity{Account: u.Account, App: u.App} |
| + ok, reason := s.adm.Admit(id) |
| + if !ok { |
| + s.countRefusal(id, reason) |
| + s.log("relay: allocation refused for %s/%s: %s", id.Account, id.App, reason) |
| + return false |
| + } |
| + key := typedSrc(src) |
| + s.mu.Lock() |
| + old, had := s.admits[key] |
| + s.admits[key] = admitRec{id: id, at: now} |
| + s.mu.Unlock() |
| + if had { |
| + // A previous attempt from this source was admitted and never |
| + // became an allocation; it owns no socket, so release it here. |
| + s.adm.Release(old.id) |
| + } |
| + return true |
| +} |
| + |
| +// created is pion's OnAllocationCreated, called with the relay address |
| +// before pion starts reading from the socket. It is where the wrapper we |
| +// handed out earlier learns whose allocation it is. |
| +func (s *Server) created(src, _ net.Addr, _, _, _ string, relayAddr net.Addr, _ int) { |
| + ua, ok := relayAddr.(*net.UDPAddr) |
| + if !ok { |
| + s.log("relay: allocation created with a non-UDP relay address %v", relayAddr) |
| + return |
| + } |
| + key := typedSrc(src) |
| + now := s.now() |
| + |
| + s.mu.Lock() |
| + p, havePending := s.pending[ua.Port] |
| + if !havePending { |
| + s.mu.Unlock() |
| + s.log("relay: allocation created on port %d with no socket of ours", ua.Port) |
| + return |
| + } |
| + delete(s.pending, ua.Port) |
| + rec, haveAuth := s.lastAuth[key] |
| + if !haveAuth { |
| + s.mu.Unlock() |
| + s.log("relay: allocation created for %s with no authenticated username", key) |
| + p.conn.Close() |
| + return |
| + } |
| + c := p.conn |
| + c.b.ID = Identity{Account: rec.u.Account, App: rec.u.App} |
| + c.b.Expiry = rec.u.Expiry |
| + c.b.Kid = rec.u.Kid |
| + 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} |
| + s.allocs[key] = e |
| + s.owners[c] = e |
| + s.bound[key] = c.b |
| + delete(s.admits, key) |
| + expiry := rec.u.Expiry |
| + s.mu.Unlock() |
| + |
| + // The credential's own deadline. Set outside the lock: a timer whose |
| + // deadline has already passed fires on the spot, and closing takes the |
| + // lock we would still be holding. |
| + t := time.AfterFunc(expiry.Sub(now), func() { c.Close() }) |
| + s.mu.Lock() |
| + if cur, ok := s.allocs[key]; ok && cur == e { |
| + e.timer = t |
| + } else { |
| + t.Stop() |
| + } |
| + s.mu.Unlock() |
| + s.log("relay: allocation for %s/%s on %s from %s until %s", |
| + c.b.ID.Account, c.b.ID.App, relayAddr, key, expiry.UTC().Format(time.RFC3339)) |
| +} |
| + |
| +// release is the wrapped socket's onClose, whoever closed it: pion on a |
| +// read error, a timer, tick, EndWhere. Its state is keyed on this socket, |
| +// not on the typed source, so a replacement allocation bound for the same |
| +// source while this one was closing (pion removes an allocation from its |
| +// manager before closing it) keeps its binding and this one still gives |
| +// its admission back. |
| +func (s *Server) release(port int, c *relayConn) { |
| + s.ports.remove(port) |
| + var id Identity |
| + var release bool |
| + s.mu.Lock() |
| + if p, ok := s.pending[port]; ok && p.conn == c { |
| + delete(s.pending, port) |
| + } |
| + if e, ok := s.owners[c]; ok { |
| + delete(s.owners, c) |
| + if e.timer != nil { |
| + e.timer.Stop() |
| + } |
| + if cur, ok := s.allocs[e.key]; ok && cur == e { |
| + delete(s.allocs, e.key) |
| + } |
| + if b, ok := s.bound[e.key]; ok && b == c.b { |
| + delete(s.bound, e.key) |
| + } |
| + id, release = e.id, true |
| + } |
| + s.mu.Unlock() |
| + if release { |
| + s.adm.Release(id) |
| + } |
| +} |
| + |
| +// permit is pion's PermissionHandler: it sees only a peer IP, so the port |
| +// half of the EIP rule lives in relayConn.WriteTo. |
| +func (s *Server) permit(client net.Addr, peer net.IP) bool { |
| + deny := func(why string) bool { |
| + s.mu.Lock() |
| + b, ok := s.bound[typedSrc(client)] |
| + var id Identity |
| + if ok { |
| + id = b.ID |
| + } |
| + s.mu.Unlock() |
| + s.countRefusal(id, "peer denied") |
| + s.log("relay: permission refused for %s to %s: %s", client, peer, why) |
| + return false |
| + } |
| + switch { |
| + case peer == nil || peer.IsUnspecified(): |
| + return deny("unspecified address") |
| + case peer.IsMulticast(): |
| + return deny("multicast address") |
| + } |
| + // Our own EIP is allowed: two clients that both relay here hold relay |
| + // candidates on it and must reach each other. |
| + if s.cfg.RelayIP != nil && peer.Equal(s.cfg.RelayIP) { |
| + return true |
| + } |
| + for _, n := range s.deny { |
| + if n.Contains(peer) { |
| + return deny("denied range " + n.String()) |
| + } |
| + } |
| + for _, ip := range s.local { |
| + if peer.Equal(ip) { |
| + return deny("the relay's own address") |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// --- metering ------------------------------------------------------------ |
| + |
| +// countIn is the client-to-peer leg, counted at the relay socket after |
| +// pion forwarded it. |
| +func (s *Server) countIn(id Identity, n int) { |
| + if s.cfg.Meter != nil { |
| + s.cfg.Meter.Add(id, int64(n), 0, 0) |
| + } |
| + if s.adm.AddBytes(id, int64(n)) { |
| + s.overBudget(id) |
| + } |
| +} |
| + |
| +// countOut is the peer-to-client leg, counted at the listener socket and |
| +// attributed through the owner binding. |
| +func (s *Server) countOut(id Identity, n int) { |
| + if s.cfg.Meter != nil { |
| + s.cfg.Meter.Add(id, 0, int64(n), 0) |
| + } |
| + if s.adm.AddBytes(id, int64(n)) { |
| + s.overBudget(id) |
| + } |
| +} |
| + |
| +// 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) { |
| + if s.cfg.Meter != nil { |
| + s.cfg.Meter.Add(id, 0, 0, int64(n)) |
| + } |
| + s.countRefusal(id, "eip port") |
| +} |
| + |
| +// overBudget ends an app's allocations once its rolling-hour budget is |
| +// crossed, because a budget that only stops new allocations bounds nothing. |
| +// It runs off the forwarding path, once per identity at a time. |
| +func (s *Server) overBudget(id Identity) { |
| + s.mu.Lock() |
| + if s.shedding[id] { |
| + s.mu.Unlock() |
| + return |
| + } |
| + s.shedding[id] = true |
| + s.mu.Unlock() |
| + go func() { |
| + s.EndWhere(func(b *Binding) bool { return b.ID == id }, "byte budget") |
| + s.mu.Lock() |
| + delete(s.shedding, id) |
| + s.mu.Unlock() |
| + }() |
| +} |
| + |
| +func (s *Server) countRefusal(id Identity, reason string) { |
| + s.mu.Lock() |
| + m := s.refusals[id] |
| + if m == nil { |
| + m = map[string]int{} |
| + s.refusals[id] = m |
| + } |
| + m[reason]++ |
| + s.mu.Unlock() |
| +} |
| + |
| +// 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...) |
| + return nil, false |
| +} |
| + |
| +func (s *Server) refusalCount(id Identity, reason string) int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.refusals[id][reason] |
| +} |
| + |
| +// entryFor is the owner binding lookup the out leg is attributed through. |
| +func (s *Server) entryFor(key string) *allocEntry { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.allocs[key] |
| +} |
| + |
| +// --- listener wrappers --------------------------------------------------- |
| + |
| +// listenerConn wraps the UDP listener pion answers clients on. It mirrors |
| +// relayConn without the EIP gate: everything written here goes to a client |
| +// that asked us for it, so the only questions are whose bytes they are and |
| +// whether the buckets still have room. |
| +type listenerConn struct { |
| + net.PacketConn |
| + s *Server |
| +} |
| + |
| +func (c *listenerConn) WriteTo(p []byte, addr net.Addr) (int, error) { |
| + 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)) { |
| + return len(p), nil |
| + } |
| + e.conn.appBucket.Spend(len(p)) |
| + c.s.box.Spend(len(p)) |
| + } |
| + n, err := c.PacketConn.WriteTo(p, addr) |
| + if n > 0 && e != nil { |
| + c.s.countOut(e.id, n) |
| + } |
| + return n, err |
| +} |
| + |
| +// 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. |
| +type countingListener struct { |
| + net.Listener |
| + s *Server |
| +} |
| + |
| +func (l *countingListener) Accept() (net.Conn, error) { |
| + c, err := l.Listener.Accept() |
| + if err != nil { |
| + return nil, err |
| + } |
| + return &countingConn{Conn: c, s: l.s, key: typedSrc(c.RemoteAddr())}, nil |
| +} |
| + |
| +type countingConn struct { |
| + net.Conn |
| + s *Server |
| + key string |
| +} |
| + |
| +func (c *countingConn) Write(p []byte) (int, error) { |
| + n, err := c.Conn.Write(p) |
| + if n > 0 { |
| + if e := c.s.entryFor(c.key); e != nil { |
| + e.conn.appBucket.Spend(n) |
| + c.s.box.Spend(n) |
| + c.s.countOut(e.id, n) |
| + } |
| + } |
| + return n, err |
| +} |
| + |
| +func contains(ss []string, s string) bool { |
| + for _, x := range ss { |
| + if x == s { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
amadan