From 7da7abe73d92a41985d9c2d670c9a127ac9244da Mon Sep 17 00:00:00 2001 From: Artur Schaback Date: Tue, 4 Aug 2026 02:56:38 +0300 Subject: [PATCH 1/3] feat: outbound SIP registration (REGISTER) Adds optional per-node SIP registration, so a deployment without a stable public address can accept inbound calls from a provider that routes to a registration binding, instead of needing a registering proxy in front of it. Configured under sip_registrations: registrar, credentials, requested expiry and a NAT keepalive interval. Registration is a prerequisite for inbound calls, not a path for them, so INVITEs still arrive on the existing inbound path and are authenticated and dispatched by trunk as before. Covers digest auth on 401 and 407, refresh ahead of expiry, 423 Min-Expires negotiation, un-REGISTER on shutdown, backoff on failure, rport on the Via so a registrar can answer the address it observes behind NAT, and an OPTIONS keepalive to hold that NAT mapping open between refreshes. Refs #524 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018vEyHdjPUV3BrxxN2m41pT --- README.md | 35 ++ cmd/livekit-sip/main.go | 1 + pkg/config/config.go | 81 ++++ pkg/config/config_test.go | 59 +++ pkg/service/service.go | 12 + pkg/sip/register.go | 710 +++++++++++++++++++++++++++++ pkg/sip/register_test.go | 934 ++++++++++++++++++++++++++++++++++++++ pkg/sip/service.go | 74 +++ pkg/stats/monitor.go | 19 + test/cloud/service.go | 5 +- 10 files changed, 1929 insertions(+), 1 deletion(-) create mode 100644 pkg/config/config_test.go create mode 100644 pkg/sip/register.go create mode 100644 pkg/sip/register_test.go diff --git a/README.md b/README.md index faed8dfa5..c3c986051 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ Currently, the following features are supported: - Dialing Out (Sending INVITEs) - Dialing In (Accepting INVITEs) - Digest Authentication +- Registering with a provider (Sending REGISTERs) - Touch Tone (Sending and Reading DTMF) ## Documentation @@ -66,6 +67,40 @@ rtp_port: port to listen and send RTP traffic (default 10000-20000) The config file can be added to a mounted volume with its location passed in the SIP_CONFIG_FILE env var, or its body can be passed in the SIP_CONFIG_BODY env var. +### Registering with a provider + +Providers usually deliver inbound calls to an address you give them, which requires the SIP +service to be reachable at a stable IP. Where that is not possible, most of them offer a +registration-based trunk instead: the service registers, and calls are delivered to the +resulting binding. Configure one entry per account: + +```yaml +sip_registrations: + - registrar: sip:sip.provider.example # registrar address, or host[:port] + username: 1000 # address-of-record user + password: secret + # auth_username: 1000 # digest username, if issued separately + # domain: provider.example # AOR host, defaults to the registrar host + # expiry: 10m # requested lifetime; the registrar may grant less + # keepalive: 25s # OPTIONS interval to hold the NAT binding open +``` + +Each node registers its own Contact, so point one node at a given account. Registration only +tells the provider where to send calls: inbound INVITEs still arrive on the normal inbound path +and are authenticated and dispatched by trunk as usual. + +The Contact is this node's signaling address, which behind NAT is a private one. That works +with providers that route to the source address they observe, which is what the REGISTER asks +them to do. If yours routes to the Contact instead, set `nat_1_to_1_ip` to the address it +should use. + +`keepalive` matters because providers commonly refuse a registration interval short enough to +refresh a NAT mapping on its own, and a mapping that expires stops inbound calls while the +registration still looks healthy. Set it to a negative value to disable. + +Because losing a registration silently stops inbound calls, each one is also exported as +`livekit_sip_registrations_active`, labeled by registrar. + ### Using the SIP service #### Creating Bridge and Dispatch Rule diff --git a/cmd/livekit-sip/main.go b/cmd/livekit-sip/main.go index 7f3bd27a5..f16e08ab9 100644 --- a/cmd/livekit-sip/main.go +++ b/cmd/livekit-sip/main.go @@ -104,6 +104,7 @@ func runService(ctx context.Context, c *cli.Command) error { return err } svc := service.NewService(conf, log, sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon) + svc.SetSIPServiceDrain(sipsrv.StopRegistrations) sipsrv.SetHandler(svc) if err = sipsrv.Start(); err != nil { diff --git a/pkg/config/config.go b/pkg/config/config.go index 8192660d3..b12706db2 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -51,6 +51,52 @@ const ( DefaultRTPDrainingDuration = 10 * time.Minute // hard cap on how long a port stays draining ) +const ( + // DefaultSIPRegistrationExpiry is the registration lifetime requested from the registrar. + // The registrar may grant less, or demand more with 423 Interval Too Brief. + DefaultSIPRegistrationExpiry = 10 * time.Minute + // DefaultSIPRegistrationKeepalive is how often an OPTIONS request is sent to the registrar + // to keep the NAT binding open. NAT UDP mappings commonly expire after 30s, and providers + // often refuse a registration interval short enough to refresh the mapping on its own. + DefaultSIPRegistrationKeepalive = 25 * time.Second + // MinSIPRegistrationExpiry is the shortest registration lifetime we allow to be configured. + MinSIPRegistrationExpiry = 30 * time.Second + // MaxSIPRegistrationExpiry is the longest registration lifetime, both as configured and as + // a registrar can push us to with 423 Interval Too Brief. + MaxSIPRegistrationExpiry = 24 * time.Hour + // MinSIPRegistrationKeepalive is the shortest keepalive interval we allow to be configured. + MinSIPRegistrationKeepalive = 5 * time.Second +) + +// SIPRegistrationConfig makes this node register with a SIP registrar (RFC 3261, section 10), +// so that a provider can deliver inbound calls to a node it cannot address by a static IP. +// +// Registrations are per-node: each node that has them configured registers its own Contact. +// Point exactly one node at a given registrar account unless the provider supports (and you +// want) parallel forking to several contacts. +type SIPRegistrationConfig struct { + // Registrar is the address of the registrar: "sip:host[:port][;transport=tcp]", or just + // "host[:port]". It becomes the Request-URI of the REGISTER, and the destination it is + // sent to. + Registrar string `yaml:"registrar"` + // Username is the user part of the address-of-record to register, and the digest username + // unless AuthUsername is set. + Username string `yaml:"username"` + // Password is the digest password. May be empty if the registrar does not challenge. + Password string `yaml:"password"` + // AuthUsername overrides the digest username, for providers that issue one separately + // from the address-of-record. + AuthUsername string `yaml:"auth_username"` + // Domain is the host part of the address-of-record. Defaults to the registrar host, which + // is what most providers expect. + Domain string `yaml:"domain"` + // Expiry is the registration lifetime to request. Defaults to DefaultSIPRegistrationExpiry. + Expiry time.Duration `yaml:"expiry"` + // Keepalive is how often to send OPTIONS to the registrar to keep a NAT binding open. + // Defaults to DefaultSIPRegistrationKeepalive. Set to a negative value to disable. + Keepalive time.Duration `yaml:"keepalive"` +} + type TLSCert struct { CertFile string `yaml:"cert_file"` KeyFile string `yaml:"key_file"` @@ -103,6 +149,9 @@ type Config struct { MaxActiveCalls int `yaml:"max_active_calls"` // if set, used for affinity-based routing SIPTrunkIds []string `yaml:"sip_trunk_ids"` // if set, only accept calls for these trunk IDs + // SIPRegistrations makes this node register with the listed SIP registrars on startup. + SIPRegistrations []SIPRegistrationConfig `yaml:"sip_registrations"` + UseExternalIP bool `yaml:"use_external_ip"` LocalNet string `yaml:"local_net"` // local IP net to use, e.g. 192.168.0.0/24 NAT1To1IP string `yaml:"nat_1_to_1_ip"` @@ -208,6 +257,9 @@ func (c *Config) Init() error { if c.MaxCpuUtilization <= 0 || c.MaxCpuUtilization > 1 { c.MaxCpuUtilization = 0.9 } + if err := c.initRegistrations(); err != nil { + return err + } if err := c.InitLogger(); err != nil { return err @@ -224,6 +276,35 @@ func (c *Config) Init() error { return nil } +func (c *Config) initRegistrations() error { + for i := range c.SIPRegistrations { + r := &c.SIPRegistrations[i] + if r.Registrar == "" { + return fmt.Errorf("sip_registrations[%d]: registrar must be set", i) + } + if r.Username == "" { + return fmt.Errorf("sip_registrations[%d]: username must be set", i) + } + switch { + case r.Expiry == 0: + r.Expiry = DefaultSIPRegistrationExpiry + case r.Expiry < MinSIPRegistrationExpiry: + return fmt.Errorf("sip_registrations[%d]: expiry must be at least %s", i, MinSIPRegistrationExpiry) + case r.Expiry > MaxSIPRegistrationExpiry: + return fmt.Errorf("sip_registrations[%d]: expiry must be at most %s", i, MaxSIPRegistrationExpiry) + } + switch { + case r.Keepalive == 0: + r.Keepalive = DefaultSIPRegistrationKeepalive + case r.Keepalive < 0: + r.Keepalive = 0 // disabled + case r.Keepalive < MinSIPRegistrationKeepalive: + return fmt.Errorf("sip_registrations[%d]: keepalive must be at least %s, or negative to disable", i, MinSIPRegistrationKeepalive) + } + } + return nil +} + func (c *Config) InitLogger(values ...interface{}) error { zl, err := logger.NewZapLogger(&c.Logging) if err != nil { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go new file mode 100644 index 000000000..bd5a1c926 --- /dev/null +++ b/pkg/config/config_test.go @@ -0,0 +1,59 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 config + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestInitRegistrations(t *testing.T) { + newConf := func(regs ...SIPRegistrationConfig) *Config { + return &Config{SIPRegistrations: regs} + } + + t.Run("defaults", func(t *testing.T) { + c := newConf(SIPRegistrationConfig{Registrar: "sip.example.com", Username: "alice"}) + require.NoError(t, c.Init()) + require.Equal(t, DefaultSIPRegistrationExpiry, c.SIPRegistrations[0].Expiry) + require.Equal(t, DefaultSIPRegistrationKeepalive, c.SIPRegistrations[0].Keepalive) + }) + t.Run("keepalive disabled", func(t *testing.T) { + c := newConf(SIPRegistrationConfig{Registrar: "sip.example.com", Username: "alice", Keepalive: -1}) + require.NoError(t, c.Init()) + require.Zero(t, c.SIPRegistrations[0].Keepalive) + }) + t.Run("registrar required", func(t *testing.T) { + require.Error(t, newConf(SIPRegistrationConfig{Username: "alice"}).Init()) + }) + t.Run("username required", func(t *testing.T) { + require.Error(t, newConf(SIPRegistrationConfig{Registrar: "sip.example.com"}).Init()) + }) + t.Run("expiry out of range", func(t *testing.T) { + require.Error(t, newConf(SIPRegistrationConfig{ + Registrar: "sip.example.com", Username: "alice", Expiry: time.Second, + }).Init()) + require.Error(t, newConf(SIPRegistrationConfig{ + Registrar: "sip.example.com", Username: "alice", Expiry: 48 * time.Hour, + }).Init()) + }) + t.Run("keepalive too short", func(t *testing.T) { + require.Error(t, newConf(SIPRegistrationConfig{ + Registrar: "sip.example.com", Username: "alice", Keepalive: time.Second, + }).Init()) + }) +} diff --git a/pkg/service/service.go b/pkg/service/service.go index dbf61253b..3011c4096 100644 --- a/pkg/service/service.go +++ b/pkg/service/service.go @@ -43,6 +43,7 @@ import ( type sipServiceStopFunc func() type sipServiceActiveCallsFunc func() sip.ActiveCalls +type sipServiceDrainFunc func() type Service struct { conf *config.Config @@ -59,12 +60,19 @@ type Service struct { sipServiceStop sipServiceStopFunc sipServiceActiveCalls sipServiceActiveCallsFunc + sipServiceDrain sipServiceDrainFunc mon *stats.Monitor shutdown core.Fuse killed atomic.Bool } +// SetSIPServiceDrain registers a hook run once shutdown starts and before the service waits +// for active calls to finish, for work that should stop new calls from arriving. +func (s *Service) SetSIPServiceDrain(fn sipServiceDrainFunc) { + s.sipServiceDrain = fn +} + func NewService( conf *config.Config, log logger.Logger, srv rpc.SIPInternalServerImpl, sipServiceStop sipServiceStopFunc, sipServiceActiveCalls sipServiceActiveCallsFunc, cli rpc.IOInfoSIPClient, bus psrpc.MessageBus, mon *stats.Monitor, @@ -185,6 +193,10 @@ func (s *Service) Run() error { <-s.shutdown.Watch() s.log.Infow("shutting down") s.DeregisterCreateSIPParticipantTopic() + if s.sipServiceDrain != nil { + // Stop attracting new inbound calls before waiting for the current ones to finish. + s.sipServiceDrain() + } if !s.killed.Load() { shutdownTicker := time.NewTicker(5 * time.Second) diff --git a/pkg/sip/register.go b/pkg/sip/register.go new file mode 100644 index 000000000..04d6d9c8e --- /dev/null +++ b/pkg/sip/register.go @@ -0,0 +1,710 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 sip + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "sync/atomic" + "time" + + "github.com/frostbyte73/core" + "github.com/icholy/digest" + + esip "github.com/emiago/sipgo/sip" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/sipgo/sip" + "github.com/livekit/sipgo/transport" + + "github.com/livekit/sip/pkg/config" + "github.com/livekit/sip/pkg/stats" +) + +const ( + // registerMaxAttempts bounds the REGISTER transactions sent for one refresh. A refresh + // needs one request, plus one retry per challenge and one per expiry renegotiation. + registerMaxAttempts = 5 + + registerMinBackoff = 2 * time.Second + registerMaxBackoff = 2 * time.Minute + + // registerTimeout bounds one REGISTER exchange, including its authentication retry. A + // non-INVITE client transaction dies at timer B after 32s anyway, and a registrar that + // did not answer the first request is better handled by the retry loop than by waiting. + registerTimeout = 32 * time.Second + // unregisterTimeout bounds the un-REGISTER we send on shutdown, which delays it. + unregisterTimeout = 5 * time.Second + // keepaliveTimeout bounds one OPTIONS keepalive exchange. Keepalives share the goroutine + // that refreshes the registration, so this must stay well below registerMinRefreshMargin: + // a registrar that stops answering must not be able to delay a refresh past the expiry. + keepaliveTimeout = 3 * time.Second + + // A refresh is sent this long before the registration expires, so a failure leaves + // room for a retry while the current binding is still valid. + registerMinRefreshMargin = 10 * time.Second + registerMaxRefreshMargin = time.Minute + // registerMinRefreshInterval keeps a peer that grants an absurdly short lifetime from + // turning refreshes into a hot loop. + registerMinRefreshInterval = time.Second + + // signalingSocketTimeout bounds the wait for the SIP server's UDP listener at startup. + signalingSocketTimeout = 10 * time.Second +) + +// errAuthFailed marks an authentication failure that retrying cannot fix on its own: the +// credentials have to change, here or at the provider. +var errAuthFailed = errors.New("registration credentials were not accepted") + +// RegistrationState is a snapshot of one registration. +type RegistrationState struct { + // Registered reports that the registrar has a binding for us. It stays true while a + // refresh is failing, because the binding that refresh would replace is still live until + // its granted lifetime runs out, and false once it does. + Registered bool + // Expiry is the lifetime the registrar granted for the current binding. + Expiry time.Duration + // Error is the failure that ended the last attempt, if it failed. + Error error +} + +// registrant keeps a single address-of-record registered with a remote registrar, so that a +// provider can deliver inbound calls without addressing this node directly (RFC 3261, §10). +// +// Registration is a prerequisite for inbound calls, not a path for them: once the provider has +// a binding, its INVITEs arrive on the normal inbound path and are authenticated and dispatched +// by trunk as usual. +type registrant struct { + log logger.Logger + cli SIPClient + mon *stats.Monitor + + // registrar is the Request-URI of the REGISTER: the registrar's own address, no user part. + registrar sip.Uri + // aor is the address-of-record being registered, used for From and To. + aor sip.Uri + // contact is the address we ask the registrar to bind the AOR to. + contact sip.Uri + // transport is the SIP transport used to reach the registrar. + transport string + // viaHost is the host this node puts in Via, matching what the SIP client announces. + viaHost string + + authUser string + password string + keepalive time.Duration + + // callID stays constant for the lifetime of the registration and cseq only increases, so + // that the registrar can order our requests for this AOR (RFC 3261, §10.2). + callID string + fromTag string + kaCallID string + kaFromTag string + + // Fields below are owned by run() and must not be touched from other goroutines. + cseq uint32 + kaCSeq uint32 + // expiry is the lifetime we ask for, raised in place when a registrar demands more. + expiry time.Duration + // challenge is the last challenge we were given. One slot is enough for the providers + // this targets: a peer that wants both a 401 and a 407 answered on the same request is + // not supported. + challenge *digest.Challenge + // authHeader is the request header that answers the cached challenge: "Authorization" + // for a 401, "Proxy-Authorization" for a 407. + authHeader string + nonceCount int + observed string + // bound records that the registrar has accepted a binding that has not been removed, so + // shutdown still withdraws it after a refresh failed. A binding left pointing at a stopped + // node sends inbound calls nowhere until it expires on its own. + bound bool + // granted is the lifetime of the current binding, and boundUntil when it lapses. Past + // that point the binding is gone whatever we last heard, so we stop claiming it. + granted time.Duration + boundUntil time.Time + + state atomic.Pointer[RegistrationState] + started atomic.Bool + + stop core.Fuse + done chan struct{} +} + +func newRegistrant(log logger.Logger, cli SIPClient, mon *stats.Monitor, conf *config.Config, sconf *ServiceConfig, rc config.SIPRegistrationConfig) (*registrant, error) { + registrar, err := parseRegistrarURI(rc.Registrar) + if err != nil { + return nil, err + } + tr := registrarTransport(®istrar) + + domain := rc.Domain + if domain == "" { + domain = registrar.Host + } + aor := sip.Uri{Scheme: registrar.Scheme, User: rc.Username, Host: domain} + + contact := getContactURI(conf, sconf.SignalingIP, tr) + contact.User = rc.Username + + authUser := rc.AuthUsername + if authUser == "" { + authUser = rc.Username + } + expiry := rc.Expiry + if expiry <= 0 { + // Config.Init sets this; a zero here would ask for a binding of no length at all. + expiry = config.DefaultSIPRegistrationExpiry + } + + r := ®istrant{ + cli: cli, + mon: mon, + registrar: registrar, + aor: aor, + contact: *contact.GetContactURI(), + transport: strings.ToUpper(string(tr)), + viaHost: sconf.SignalingIP.String(), + authUser: authUser, + password: rc.Password, + expiry: expiry, + keepalive: rc.Keepalive, + callID: sip.GenerateTagN(32), + fromTag: sip.GenerateTagN(16), + kaCallID: sip.GenerateTagN(32), + kaFromTag: sip.GenerateTagN(16), + done: make(chan struct{}), + } + r.log = log.WithValues( + "registrar", r.registrar.String(), + "aor", r.aor.String(), + "contact", r.contact.String(), + ) + r.setState(RegistrationState{}) + return r, nil +} + +// parseRegistrarURI accepts a SIP URI or a bare "host[:port]" address. +func parseRegistrarURI(s string) (sip.Uri, error) { + if !strings.HasPrefix(s, "sip:") && !strings.HasPrefix(s, "sips:") { + s = "sip:" + s + } + var u sip.Uri + if err := esip.ParseUri(s, &u); err != nil { + return sip.Uri{}, fmt.Errorf("invalid registrar %q: %w", s, err) + } + if u.Host == "" { + return sip.Uri{}, fmt.Errorf("invalid registrar %q: no host", s) + } + // The Request-URI of a REGISTER names the registrar, never a user on it. + u.User, u.Password = "", "" + return u, nil +} + +// registrarTransport reports the SIP transport to reach a registrar with. sipgo only derives +// TLS from a sips: URI when a transport parameter already says TCP, so resolve it here and set +// it on every request instead. +func registrarTransport(u *sip.Uri) Transport { + if u.IsEncrypted() { + return TransportTLS + } + if t := transportFromURI(u); t != "" { + return t + } + return TransportUDP +} + +// dest is the transport address requests to the registrar are sent to. +func (r *registrant) dest() string { + return r.registrar.Host + ":" + strconv.Itoa(uriPort(&r.registrar)) +} + +func (r *registrant) Start() { + if !r.started.CompareAndSwap(false, true) { + return + } + go r.run() +} + +// Stop unregisters and waits for the registration to shut down. It must be called before the +// SIP client is closed, since the un-REGISTER is sent over it. +func (r *registrant) Stop() { + r.stop.Break() + if r.started.Load() { + <-r.done + } +} + +func (r *registrant) State() RegistrationState { + return *r.state.Load() +} + +func (r *registrant) setState(st RegistrationState) { + r.state.Store(&st) + if r.mon != nil { + r.mon.RegistrationActive(r.registrar.String(), st.Registered) + } +} + +func (r *registrant) run() { + defer close(r.done) + + // Fires immediately for the initial registration, then rescheduled per response. + refresh := time.NewTimer(0) + defer refresh.Stop() + + // Fires when the current binding lapses, so a long backoff cannot leave us reporting one + // that is already gone. Every successful refresh pushes it out again. + lapse := time.NewTimer(0) + <-lapse.C + defer lapse.Stop() + + var keepalive <-chan time.Time + if r.keepalive > 0 { + t := time.NewTicker(r.keepalive) + defer t.Stop() + keepalive = t.C + } + + backoff := registerMinBackoff + for { + select { + case <-r.stop.Watch(): + r.unregister() + return + case <-refresh.C: + granted, err := r.registerOnce(r.refreshTimeout(), r.expiry) + if err == nil { + // Record the binding before anything else can return: a REGISTER answered as + // shutdown starts still created one, and the stop branch has to withdraw it. + r.bound, r.granted, r.boundUntil = true, granted, time.Now().Add(granted) + } + if r.stop.IsBroken() { + continue // shutting down; the stop branch unregisters + } + if err != nil { + // A binding that has not expired yet is still live, so keep reporting it and + // keep the keepalives going: they hold open the NAT mapping the provider is + // already sending calls to. Once it lapses we stop claiming it. + r.setState(RegistrationState{Registered: r.stillBound(), Expiry: r.granted, Error: err}) + if errors.Is(err, errAuthFailed) { + // Nothing here is fixed by trying again soon; the account has to change. + backoff = registerMaxBackoff + r.log.Errorw("SIP registration was rejected", err, "retryIn", backoff) + } else { + r.log.Warnw("SIP registration failed, retrying", err, "retryIn", backoff) + } + refresh.Reset(backoff) + if backoff *= 2; backoff > registerMaxBackoff { + backoff = registerMaxBackoff + } + continue + } + backoff = registerMinBackoff + after := refreshAfter(granted) + if r.State().Registered { + r.log.Debugw("SIP registration refreshed", "expiry", granted, "refreshIn", after) + } else { + r.log.Infow("SIP registration established", "expiry", granted, "refreshIn", after) + } + r.setState(RegistrationState{Registered: true, Expiry: granted}) + refresh.Reset(after) + lapse.Reset(granted) + case <-lapse.C: + if r.stop.IsBroken() || r.stillBound() { + continue // refreshed in the meantime + } + last := r.State() + r.bound, r.granted, r.boundUntil = false, 0, time.Time{} + r.setState(RegistrationState{Error: last.Error}) + r.log.Warnw("SIP registration lapsed; inbound calls will not arrive", last.Error) + case <-keepalive: + if r.stop.IsBroken() || !r.stillBound() { + continue // shutting down, or no binding to hold open + } + r.sendKeepalive() + } + } +} + +// stillBound reports whether the registrar should still have a binding for us. +func (r *registrant) stillBound() bool { + return r.bound && time.Now().Before(r.boundUntil) +} + +// refreshTimeout bounds one refresh attempt: there is little point waiting out the full +// transaction timeout past the moment the current binding lapses, when failing sooner puts us +// back in the retry loop while there is still time to replace it. The floor keeps a nearly +// lapsed binding from cutting the attempt uselessly short, since a late 200 still rebinds. +func (r *registrant) refreshTimeout() time.Duration { + timeout := registerTimeout + if remaining := time.Until(r.boundUntil); r.bound && remaining > 0 && remaining < timeout { + timeout = remaining + } + if timeout < registerMinRefreshMargin { + timeout = registerMinRefreshMargin + } + return timeout +} + +// refreshAfter returns how long to wait before refreshing a registration granted for d. +func refreshAfter(d time.Duration) time.Duration { + margin := d / 10 + if margin < registerMinRefreshMargin { + margin = registerMinRefreshMargin + } else if margin > registerMaxRefreshMargin { + margin = registerMaxRefreshMargin + } + after := d - margin + if d <= margin { + after = d / 2 + } + if after < registerMinRefreshInterval { + after = registerMinRefreshInterval + } + return after +} + +// registerOnce runs one REGISTER exchange, answering an auth challenge and renegotiating the +// expiry if the registrar demands a longer one, and returns the granted lifetime. An expires +// of 0 removes the binding. +func (r *registrant) registerOnce(timeout time.Duration, expires time.Duration) (time.Duration, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + // The un-REGISTER on shutdown must survive the stop signal that triggered it. + var stop <-chan struct{} + if expires > 0 { + stop = r.stop.Watch() + } + + minExpiresApplied := false + for attempt := 0; attempt < registerMaxAttempts; attempt++ { + req := r.newRequest(sip.REGISTER, r.callID, r.fromTag, &r.cseq) + req.AppendHeader(&sip.ContactHeader{Address: cloneURI(r.contact)}) + exp := sip.ExpiresHeader(expires / time.Second) + req.AppendHeader(&exp) + if r.challenge != nil { + if err := r.authorize(req); err != nil { + return 0, err + } + } + + resp, err := r.roundtrip(ctx, stop, req) + if err != nil { + return 0, err + } + switch resp.StatusCode { + case sip.StatusOK: + r.logObservedAddress(resp) + granted := grantedExpiry(resp, &r.contact, expires) + if expires > 0 && granted <= 0 { + // A 200 that grants nothing means the binding was not created after all. + return 0, errors.New("registrar accepted the REGISTER but granted no binding") + } + return granted, nil + case sip.StatusUnauthorized, sip.StatusProxyAuthRequired: + if err := r.setChallenge(resp); err != nil { + return 0, err + } + case sip.StatusForbidden: + // Registrars commonly answer bad credentials with 403 rather than another 401. + return 0, fmt.Errorf("%w: %w", errAuthFailed, sipStatusError(resp)) + case sip.StatusIntervalToBrief: + // The registrar demands a longer lifetime than we asked for. + if expires == 0 || minExpiresApplied { + return 0, sipStatusError(resp) + } + minExpires, ok := headerSeconds(resp, "Min-Expires") + if !ok || minExpires <= expires || minExpires > config.MaxSIPRegistrationExpiry { + return 0, fmt.Errorf("registrar demands an unusable expiry: %w", sipStatusError(resp)) + } + r.log.Infow("registrar requires a longer expiry", "requested", expires, "minExpires", minExpires) + expires, minExpiresApplied = minExpires, true + // Ask for it directly from now on, instead of being refused once per refresh. + r.expiry = minExpires + default: + return 0, sipStatusError(resp) + } + } + return 0, fmt.Errorf("REGISTER did not complete in %d attempts", registerMaxAttempts) +} + +func (r *registrant) unregister() { + if !r.stillBound() { + return // nothing bound, or it has lapsed on its own + } + r.bound, r.granted, r.boundUntil = false, 0, time.Time{} + if _, err := r.registerOnce(unregisterTimeout, 0); err != nil { + r.log.Warnw("could not unregister", err) + } else { + r.log.Infow("SIP registration removed") + } + r.setState(RegistrationState{}) +} + +// sendKeepalive sends an OPTIONS request to the registrar. A registration lifetime long enough +// for a provider to accept (often 10 minutes or more) far outlives a NAT UDP mapping, which is +// commonly dropped after 30s of silence. Without traffic on the signaling socket the binding +// the provider recorded stops being reachable, and inbound calls quietly stop arriving while +// the registration still looks healthy. Any response, including an error, keeps the mapping +// open and proves the path is still there. +func (r *registrant) sendKeepalive() { + ctx, cancel := context.WithTimeout(context.Background(), keepaliveTimeout) + defer cancel() + + req := r.newRequest(sip.OPTIONS, r.kaCallID, r.kaFromTag, &r.kaCSeq) + resp, err := r.roundtrip(ctx, r.stop.Watch(), req) + if err != nil { + if !r.stop.IsBroken() { + r.log.Warnw("no response to SIP keepalive; inbound calls may not reach this node", err) + } + return + } + r.log.Debugw("SIP keepalive answered", "status", resp.StatusCode) + r.logObservedAddress(resp) +} + +// newRequest builds an out-of-dialog request towards the registrar. +func (r *registrant) newRequest(method sip.RequestMethod, callID, fromTag string, cseq *uint32) *sip.Request { + req := sip.NewRequest(method, r.registrar) + req.SetTransport(r.transport) + + // sipgo only adds a Via of its own when the request has none, and the one it adds has no + // rport. Behind NAT our Via carries an address the registrar cannot reply to, so it must + // ask the registrar to answer the source address and port it actually observed + // (RFC 3581); without it the response never arrives and registration never completes. + via := &sip.ViaHeader{ + ProtocolName: "SIP", + ProtocolVersion: "2.0", + Transport: r.transport, + Host: r.viaHost, + Params: sip.NewParams(), + } + via.Params.Add("branch", sip.GenerateBranchN(16)) + via.Params.Add("rport", "") + req.AppendHeader(via) + + from := &sip.FromHeader{Address: cloneURI(r.aor), Params: sip.NewParams()} + from.Params.Add("tag", fromTag) + req.AppendHeader(from) + req.AppendHeader(&sip.ToHeader{Address: cloneURI(r.aor), Params: sip.NewParams()}) + + cid := sip.CallIDHeader(callID) + req.AppendHeader(&cid) + + *cseq++ + req.AppendHeader(&sip.CSeqHeader{MethodName: method, SeqNo: *cseq}) + + req.AppendHeader(sip.NewHeader("User-Agent", UserAgent)) + return req +} + +func (r *registrant) roundtrip(ctx context.Context, stop <-chan struct{}, req *sip.Request) (*sip.Response, error) { + tx, err := r.cli.TransactionRequest(req) + if err != nil { + return nil, err + } + defer tx.Terminate() + return sipResponse(ctx, tx, stop, nil) +} + +// setChallenge stores the challenge from a 401 or 407 so the next request can answer it. +func (r *registrant) setChallenge(resp *sip.Response) error { + name, authHeader := "WWW-Authenticate", "Authorization" + if resp.StatusCode == sip.StatusProxyAuthRequired { + name, authHeader = "Proxy-Authenticate", "Proxy-Authorization" + } + h := resp.GetHeader(name) + if h == nil { + return fmt.Errorf("%w: %s", ErrAuthNoHeader, name) + } + chal, err := digest.ParseChallenge(h.Value()) + if err != nil { + return fmt.Errorf("invalid %s challenge %q: %w", name, h.Value(), err) + } + // A repeat of a nonce we already answered, without stale=true, means the credentials were + // wrong rather than expired. Retrying would send the same digest and loop. + if r.challenge != nil && r.challenge.Nonce == chal.Nonce && !chal.Stale { + r.challenge = nil + return fmt.Errorf("%w: the registrar repeated its challenge", errAuthFailed) + } + if r.password == "" { + return fmt.Errorf("%w: %w", errAuthFailed, ErrAuthMissingCreds) + } + // RFC 2617: the nonce count must keep increasing for a given nonce, so only a new one + // restarts the count. A stale challenge may repeat the nonce. + if r.challenge == nil || r.challenge.Nonce != chal.Nonce { + r.nonceCount = 0 + } + r.challenge, r.authHeader = chal, authHeader + return nil +} + +func (r *registrant) authorize(req *sip.Request) error { + r.nonceCount++ + cred, err := digest.Digest(r.challenge, digest.Options{ + Method: req.Method.String(), + // The digest URI is the Request-URI (RFC 3261, §22.4). For REGISTER that is the + // registrar, which is not the same as the address-of-record in To. + URI: req.Recipient.String(), + Username: r.authUser, + Password: r.password, + Count: r.nonceCount, + }) + if err != nil { + // Typically an algorithm icholy/digest does not implement, which no retry resolves. + return fmt.Errorf("%w: %w", errAuthFailed, err) + } + req.AppendHeader(sip.NewHeader(r.authHeader, cred.String())) + return nil +} + +// logObservedAddress reports the source address the registrar saw, which RFC 3581 asks it to +// echo in the Via. Behind NAT it is the public mapping, and it will not match our Contact. +// Registrars differ in which of the two they route inbound calls to, so surfacing the +// difference separates "registered, but the provider is calling an address we do not have" +// from "not registered". +func (r *registrant) logObservedAddress(resp *sip.Response) { + via := resp.Via() + if via == nil { + return + } + host, _ := via.Params.Get("received") + port, _ := via.Params.Get("rport") + if host == "" && port == "" { + return + } + observed := host + if port != "" { + observed += ":" + port + } + if observed == r.observed { + return + } + r.observed = observed + // For TLS the Contact carries sip_hostname, which never equals an observed IP, so there is + // nothing to compare and no advice worth giving. + if host != "" && host != r.contact.Host && r.transport != strings.ToUpper(string(TransportTLS)) { + r.log.Infow("registrar sees this node at a different address than the Contact we sent; "+ + "inbound calls only arrive if it routes to the observed address, otherwise set nat_1_to_1_ip", + "observed", observed) + return + } + r.log.Debugw("registrar confirmed our address", "observed", observed) +} + +// grantedExpiry reports the lifetime the registrar granted, preferring the most specific +// source: the expires parameter on our own binding, then the Expires header, then the shortest +// binding it listed, then the value we asked for. The third step covers registrars that rewrite +// the Contact to the address they observed, which stops it matching ours. Underestimating the +// lifetime only costs an early refresh, while overestimating lets the binding lapse. +func grantedExpiry(resp *sip.Response, contact *sip.Uri, requested time.Duration) time.Duration { + var shortest time.Duration + for _, h := range resp.GetHeaders("Contact") { + c, ok := h.(*sip.ContactHeader) + if !ok { + continue + } + v, ok := c.Params.Get("expires") + if !ok { + continue + } + sec, err := strconv.Atoi(v) + if err != nil || sec < 0 { + continue + } + d := time.Duration(sec) * time.Second + if sameContact(&c.Address, contact) { + return d + } + if d > 0 && (shortest == 0 || d < shortest) { + shortest = d + } + } + if d, ok := headerSeconds(resp, "Expires"); ok { + return d + } + if shortest != 0 { + return shortest + } + return requested +} + +func sameContact(a, b *sip.Uri) bool { + return a.User == b.User && + strings.EqualFold(a.Host, b.Host) && + uriPort(a) == uriPort(b) +} + +func uriPort(u *sip.Uri) int { + if u.Port != 0 { + return u.Port + } + if u.IsEncrypted() { + return 5061 + } + return 5060 +} + +func headerSeconds(resp *sip.Response, name string) (time.Duration, bool) { + h := resp.GetHeader(name) + if h == nil { + return 0, false + } + sec, err := strconv.Atoi(strings.TrimSpace(h.Value())) + if err != nil || sec < 0 { + return 0, false + } + return time.Duration(sec) * time.Second, true +} + +func sipStatusError(resp *sip.Response) error { + return &livekit.SIPStatus{ + Code: livekit.SIPStatusCode(resp.StatusCode), + Status: resp.Reason, + } +} + +func cloneURI(u sip.Uri) sip.Uri { + u.UriParams = u.UriParams.Clone() + u.Headers = u.Headers.Clone() + return u +} + +// waitForSignalingSocket blocks until the SIP server's UDP listener is registered with the +// transport layer. sipgo reuses that socket for outbound UDP requests, so waiting makes +// REGISTER leave from the signaling port: providers deliver inbound INVITEs to the NAT mapping +// the REGISTER created, and a mapping for some other port routes them nowhere. Sending before +// the listener exists binds a separate socket, which is then cached for the registrar address +// and used for every later request. +func waitForSignalingSocket(ctx context.Context, tp *transport.Layer, dest string) error { + poll := time.NewTicker(20 * time.Millisecond) + defer poll.Stop() + for { + if _, err := tp.GetConnection("udp", dest); err == nil { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-poll.C: + } + } +} diff --git a/pkg/sip/register_test.go b/pkg/sip/register_test.go new file mode 100644 index 000000000..724615966 --- /dev/null +++ b/pkg/sip/register_test.go @@ -0,0 +1,934 @@ +// Copyright 2026 LiveKit, Inc. +// +// 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 sip + +import ( + "context" + "fmt" + "log/slog" + "math/rand" + "net" + "net/netip" + "strconv" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/icholy/digest" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/logger" + "github.com/livekit/sipgo" + "github.com/livekit/sipgo/sip" + + "github.com/livekit/sip/pkg/config" +) + +const ( + testRegUser = "reguser" + testRegPass = "regpass" + testRegRealm = "registrar.test" +) + +// testRegistrar is a minimal registrar: it records the requests it receives and answers each +// one with the response the test queued for it. +type testRegistrar struct { + t *testing.T + addr string + + mu sync.Mutex + requests []*sip.Request + got chan struct{} + + // respond returns the response to send for a request. Called with the lock held. + respond func(req *sip.Request) *sip.Response +} + +func newTestRegistrar(t *testing.T, respond func(req *sip.Request) *sip.Response) *testRegistrar { + t.Helper() + localIP, err := config.GetLocalIP() + require.NoError(t, err) + + lis, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IP(localIP.AsSlice()), Port: 0}) + require.NoError(t, err) + + r := &testRegistrar{ + t: t, + addr: lis.LocalAddr().String(), + got: make(chan struct{}, 64), + respond: respond, + } + + log := slog.New(logger.ToSlogHandler(logger.LogRLogger(logr.Discard()))) + ua, err := sipgo.NewUA(sipgo.WithUserAgent("test-registrar"), sipgo.WithUserAgentLogger(log)) + require.NoError(t, err) + srv, err := sipgo.NewServer(ua, sipgo.WithServerLogger(log)) + require.NoError(t, err) + + handle := func(_ *slog.Logger, req *sip.Request, tx sip.ServerTransaction) { + r.mu.Lock() + r.requests = append(r.requests, req) + resp := r.respond(req) + r.mu.Unlock() + if resp != nil { + _ = tx.Respond(resp) + } + select { + case r.got <- struct{}{}: + default: + } + } + srv.OnRegister(handle) + srv.OnOptions(handle) + + go func() { + _ = srv.ServeUDP(lis) + }() + t.Cleanup(func() { + _ = srv.Close() + _ = ua.Close() + _ = lis.Close() + }) + return r +} + +// waitRequests waits until at least n requests have been received. +func (r *testRegistrar) waitRequests(n int) []*sip.Request { + r.t.Helper() + deadline := time.After(10 * time.Second) + for { + if got := r.received(); len(got) >= n { + return got + } + select { + case <-r.got: + case <-deadline: + r.t.Fatalf("timed out waiting for %d requests, got %d", n, len(r.received())) + } + } +} + +func (r *testRegistrar) received() []*sip.Request { + r.mu.Lock() + defer r.mu.Unlock() + return append([]*sip.Request(nil), r.requests...) +} + +// challenge answers a request with 401 and a digest challenge. +func challenge(req *sip.Request, qop []string) *sip.Response { + chal := digest.Challenge{ + Realm: testRegRealm, + Nonce: strconv.Itoa(rand.Int()), + Algorithm: "MD5", + QOP: qop, + } + resp := sip.NewResponseFromRequest(req, sip.StatusUnauthorized, "Unauthorized", nil) + resp.AppendHeader(sip.NewHeader("WWW-Authenticate", chal.String())) + return resp +} + +// checkCredentials verifies the digest in a request against the expected password. +func checkCredentials(t *testing.T, req *sip.Request, chal *digest.Challenge, user, pass string) { + t.Helper() + h := req.GetHeader("Authorization") + require.NotNil(t, h, "request has no Authorization header") + cred, err := digest.ParseCredentials(h.Value()) + require.NoError(t, err) + require.Equal(t, user, cred.Username) + // The digest URI must be the Request-URI, not the address-of-record in To. + require.Equal(t, req.Recipient.String(), cred.URI) + + want, err := digest.Digest(chal, digest.Options{ + Method: req.Method.String(), + URI: cred.URI, + Username: user, + Password: pass, + Cnonce: cred.Cnonce, + Count: cred.Nc, + }) + require.NoError(t, err) + require.Equal(t, want.Response, cred.Response) +} + +// newTestRegistrant builds a registrant talking to addr over a real SIP client, listening on +// its own signaling socket the way the service does. It returns the registrant and the +// signaling port, which is the port requests must be sent from. +func newTestRegistrant(t *testing.T, addr string, rc config.SIPRegistrationConfig) (*registrant, int) { + t.Helper() + localIP, err := config.GetLocalIP() + require.NoError(t, err) + + log := slog.New(logger.ToSlogHandler(logger.LogRLogger(logr.Discard()))) + ua, err := sipgo.NewUA(sipgo.WithUserAgent(UserAgent), sipgo.WithUserAgentLogger(log)) + require.NoError(t, err) + cli, err := sipgo.NewClient(ua, sipgo.WithClientHostname(localIP.String()), sipgo.WithClientLogger(log)) + require.NoError(t, err) + + // Serve the signaling socket, so sipgo sends our requests from it rather than binding a + // fresh one, as it does in the service. + lis, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IP(localIP.AsSlice()), Port: 0}) + require.NoError(t, err) + srv, err := sipgo.NewServer(ua, sipgo.WithServerLogger(log)) + require.NoError(t, err) + go func() { + _ = srv.ServeUDP(lis) + }() + t.Cleanup(func() { + _ = cli.Close() + _ = srv.Close() + _ = ua.Close() + _ = lis.Close() + }) + sipPort := lis.LocalAddr().(*net.UDPAddr).Port + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + require.NoError(t, waitForSignalingSocket(ctx, ua.TransportLayer(), addr)) + + if rc.Registrar == "" { + rc.Registrar = addr + } + if rc.Username == "" { + rc.Username = testRegUser + } + if rc.Expiry == 0 { + rc.Expiry = time.Minute + } + r, err := newRegistrant( + logger.LogRLogger(logr.Discard()), cli, nil, + &config.Config{SIPPort: sipPort}, + &ServiceConfig{SignalingIP: localIP}, + rc, + ) + require.NoError(t, err) + return r, sipPort +} + +func TestRegistrantRegisters(t *testing.T) { + var ( + mu sync.Mutex + chal *digest.Challenge + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if req.GetHeader("Authorization") == nil { + resp := challenge(req, nil) + chal, _ = digest.ParseChallenge(resp.GetHeader("WWW-Authenticate").Value()) + return resp + } + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + if c := req.Contact(); c != nil { + resp.AppendHeader(c.Clone()) + } + if e := req.GetHeader("Expires"); e != nil { + resp.AppendHeader(sip.NewHeader("Expires", e.Value())) + } + return resp + }) + + r, sipPort := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{ + Password: testRegPass, + Expiry: 90 * time.Second, + }) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(2) + first, second := reqs[0], reqs[1] + + require.Equal(t, sip.REGISTER, first.Method) + require.Nil(t, first.GetHeader("Authorization"), "first REGISTER must not be pre-authenticated") + + // RFC 3581: without rport the registrar answers the address in our Via, which behind NAT + // is not reachable. + via := first.Via() + require.NotNil(t, via) + require.True(t, via.Params.Has("rport"), "Via must request rport") + + // The REGISTER has to leave from the signaling socket: behind NAT the provider delivers + // inbound INVITEs to the mapping it created, so a mapping for any other port is useless. + _, srcPort, err := net.SplitHostPort(first.Source()) + require.NoError(t, err) + require.Equal(t, strconv.Itoa(sipPort), srcPort) + require.Equal(t, sipPort, via.Port) + + // The Request-URI names the registrar and carries no user part. + require.Equal(t, "", first.Recipient.User) + require.Equal(t, r.registrar.Host, first.Recipient.Host) + + // The address-of-record is registered, bound to our Contact. + to := first.To() + require.NotNil(t, to) + require.Equal(t, testRegUser, to.Address.User) + contact := first.Contact() + require.NotNil(t, contact) + require.Equal(t, testRegUser, contact.Address.User) + require.Equal(t, sip.ExpiresHeader(90).Value(), first.GetHeader("Expires").Value()) + + mu.Lock() + c := chal + mu.Unlock() + checkCredentials(t, second, c, testRegUser, testRegPass) + + // RFC 3261 §10.2: the same Call-ID with an increasing CSeq for one address-of-record. + require.Equal(t, first.CallID().Value(), second.CallID().Value()) + require.Equal(t, first.CSeq().SeqNo+1, second.CSeq().SeqNo) + + require.Eventually(t, func() bool { + return r.State().Registered + }, 5*time.Second, 10*time.Millisecond) + require.Equal(t, 90*time.Second, r.State().Expiry) +} + +func TestRegistrantUsesAuthUsername(t *testing.T) { + const authUser = "digest-user" + var ( + mu sync.Mutex + chal *digest.Challenge + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if req.GetHeader("Authorization") == nil { + resp := challenge(req, []string{"auth"}) + chal, _ = digest.ParseChallenge(resp.GetHeader("WWW-Authenticate").Value()) + return resp + } + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{ + AuthUsername: authUser, + Password: testRegPass, + }) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(2) + require.Equal(t, testRegUser, reqs[1].To().Address.User, "the AOR keeps the configured username") + + mu.Lock() + c := chal + mu.Unlock() + // qop=auth, so the credentials also carry a client nonce and nonce count. + checkCredentials(t, reqs[1], c, authUser, testRegPass) + cred, err := digest.ParseCredentials(reqs[1].GetHeader("Authorization").Value()) + require.NoError(t, err) + require.Equal(t, "auth", cred.QOP) + require.Equal(t, 1, cred.Nc) +} + +func TestRegistrantNegotiatesMinExpires(t *testing.T) { + const minExpires = 600 + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + exp, _ := strconv.Atoi(req.GetHeader("Expires").Value()) + if exp < minExpires { + resp := sip.NewResponseFromRequest(req, sip.StatusIntervalToBrief, "Interval Too Brief", nil) + resp.AppendHeader(sip.NewHeader("Min-Expires", strconv.Itoa(minExpires))) + return resp + } + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", strconv.Itoa(exp))) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Expiry: 60 * time.Second}) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(2) + require.Equal(t, "60", reqs[0].GetHeader("Expires").Value()) + require.Equal(t, "600", reqs[1].GetHeader("Expires").Value()) + + require.Eventually(t, func() bool { + return r.State().Registered + }, 5*time.Second, 10*time.Millisecond) + require.Equal(t, minExpires*time.Second, r.State().Expiry) +} + +func TestRegistrantRefreshesBeforeExpiry(t *testing.T) { + // The registrar grants far less than we ask for, so a refresh is due almost immediately. + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", "1")) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Expiry: time.Hour}) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(3) + require.Equal(t, reqs[0].CallID().Value(), reqs[2].CallID().Value()) + require.Greater(t, reqs[2].CSeq().SeqNo, reqs[1].CSeq().SeqNo) + require.Equal(t, time.Second, r.State().Expiry) +} + +func TestRegistrantUnregistersOnStop(t *testing.T) { + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{}) + r.Start() + reg.waitRequests(1) + require.Eventually(t, func() bool { + return r.State().Registered + }, 5*time.Second, 10*time.Millisecond) + + r.Stop() + require.False(t, r.State().Registered) + + reqs := reg.waitRequests(2) + last := reqs[len(reqs)-1] + require.Equal(t, sip.REGISTER, last.Method) + require.Equal(t, "0", last.GetHeader("Expires").Value(), "shutdown must remove the binding") + require.NotNil(t, last.Contact(), "the un-REGISTER must name our own binding, not all of them") +} + +func TestRegistrantSendsKeepalive(t *testing.T) { + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{ + Keepalive: 20 * time.Millisecond, + }) + r.Start() + t.Cleanup(r.Stop) + + var options *sip.Request + require.Eventually(t, func() bool { + for _, req := range reg.received() { + if req.Method == sip.OPTIONS { + options = req + return true + } + } + return false + }, 10*time.Second, 10*time.Millisecond, "no OPTIONS keepalive was sent") + + require.True(t, options.Via().Params.Has("rport"), "keepalive Via must request rport too") + require.Equal(t, r.registrar.Host, options.Recipient.Host) + // Keepalives share one Call-ID so the registrar sees a single stream, not a new + // transaction identity every interval. + require.Equal(t, r.kaCallID, options.CallID().Value()) + require.NotEqual(t, r.callID, options.CallID().Value()) +} + +func TestRegistrantStopsOnBadCredentials(t *testing.T) { + // Same nonce, never stale: the registrar is rejecting the password, not expiring a nonce. + chal := digest.Challenge{Realm: testRegRealm, Nonce: "fixed-nonce", Algorithm: "MD5"} + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + resp := sip.NewResponseFromRequest(req, sip.StatusUnauthorized, "Unauthorized", nil) + resp.AppendHeader(sip.NewHeader("WWW-Authenticate", chal.String())) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Password: "wrong"}) + r.Start() + t.Cleanup(r.Stop) + + // Two requests: the unauthenticated one and one answering the challenge. A third would + // mean we are looping on a nonce we already know is not accepted. + reg.waitRequests(2) + require.Eventually(t, func() bool { + return r.State().Error != nil + }, 5*time.Second, 10*time.Millisecond) + require.ErrorIs(t, r.State().Error, errAuthFailed) + require.False(t, r.State().Registered) + + time.Sleep(200 * time.Millisecond) + require.Len(t, reg.received(), 2, "backoff must keep us from hammering the registrar") +} + +// freeAddr returns an address on the local IP that nothing is listening on. +func freeAddr(t *testing.T) string { + t.Helper() + localIP, err := config.GetLocalIP() + require.NoError(t, err) + lis, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IP(localIP.AsSlice()), Port: 0}) + require.NoError(t, err) + addr := lis.LocalAddr().String() + require.NoError(t, lis.Close()) + return addr +} + +func TestRegistrantStopDuringRegister(t *testing.T) { + // Nothing answers, so a REGISTER is in flight when Stop arrives. It must abort the + // transaction rather than wait out its 32s timeout. + r, _ := newTestRegistrant(t, freeAddr(t), config.SIPRegistrationConfig{}) + r.Start() + time.Sleep(100 * time.Millisecond) + + done := make(chan struct{}) + start := time.Now() + go func() { + r.Stop() + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Stop did not return") + } + require.Less(t, time.Since(start), 2*time.Second) + require.False(t, r.State().Registered) + // Nothing was ever bound, so shutdown must not try to remove one. + require.Nil(t, r.State().Error) +} + +func TestParseRegistrarURI(t *testing.T) { + cases := []struct { + name string + in string + want string + wantErr bool + }{ + {name: "host", in: "sip.example.com", want: "sip:sip.example.com"}, + {name: "host and port", in: "sip.example.com:5070", want: "sip:sip.example.com:5070"}, + {name: "uri", in: "sip:sip.example.com", want: "sip:sip.example.com"}, + {name: "uri with transport", in: "sip:sip.example.com;transport=tcp", want: "sip:sip.example.com;transport=tcp"}, + {name: "sips", in: "sips:sip.example.com", want: "sips:sip.example.com"}, + // A REGISTER Request-URI addresses the registrar, so any user part is dropped. + {name: "user is dropped", in: "sip:alice@sip.example.com", want: "sip:sip.example.com"}, + {name: "empty", in: "", wantErr: true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + u, err := parseRegistrarURI(c.in) + if c.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + require.Equal(t, c.want, u.String()) + }) + } +} + +func TestRegistrantDefaultsAORDomainToRegistrar(t *testing.T) { + localIP, err := config.GetLocalIP() + require.NoError(t, err) + sconf := &ServiceConfig{SignalingIP: localIP} + log := logger.LogRLogger(logr.Discard()) + + r, err := newRegistrant(log, nil, nil, &config.Config{SIPPort: 5060}, sconf, config.SIPRegistrationConfig{ + Registrar: "sip:sip.example.com", + Username: "alice", + }) + require.NoError(t, err) + require.Equal(t, "sip:alice@sip.example.com", r.aor.String()) + require.Equal(t, "alice", r.authUser) + // The Contact is this node's own signaling address, so the registrar binds calls to us. + require.Equal(t, netip.AddrPortFrom(localIP, 5060).String(), fmt.Sprintf("%s:%d", r.contact.Host, r.contact.Port)) + + r, err = newRegistrant(log, nil, nil, &config.Config{SIPPort: 5060}, sconf, config.SIPRegistrationConfig{ + Registrar: "sip:sbc.example.com", + Domain: "example.com", + Username: "alice", + }) + require.NoError(t, err) + require.Equal(t, "sip:alice@example.com", r.aor.String()) + require.Equal(t, "sip:sbc.example.com", r.registrar.String()) +} + +func TestRefreshAfter(t *testing.T) { + cases := []struct { + granted time.Duration + want time.Duration + }{ + {granted: time.Second, want: registerMinRefreshInterval}, // interval floor + {granted: 10 * time.Second, want: 5 * time.Second}, // margin floor + {granted: 60 * time.Second, want: 50 * time.Second}, + {granted: 600 * time.Second, want: 540 * time.Second}, + {granted: time.Hour, want: time.Hour - time.Minute}, // margin ceiling + } + for _, c := range cases { + t.Run(c.granted.String(), func(t *testing.T) { + got := refreshAfter(c.granted) + require.Equal(t, c.want, got) + require.LessOrEqual(t, got, c.granted, "a refresh must not be scheduled past the expiry") + }) + } +} + +func TestGrantedExpiry(t *testing.T) { + contact := &sip.Uri{User: "alice", Host: "10.0.0.1", Port: 5060} + newResp := func(build func(resp *sip.Response)) *sip.Response { + req := sip.NewRequest(sip.REGISTER, sip.Uri{Host: "sip.example.com"}) + req.AppendHeader(&sip.ViaHeader{Params: sip.NewParams()}) + req.AppendHeader(&sip.FromHeader{Params: sip.NewParams()}) + req.AppendHeader(&sip.ToHeader{Params: sip.NewParams()}) + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + build(resp) + return resp + } + withContact := func(u sip.Uri, expires string) *sip.Response { + return newResp(func(resp *sip.Response) { + h := &sip.ContactHeader{Address: u, Params: sip.NewParams()} + if expires != "" { + h.Params.Add("expires", expires) + } + resp.AppendHeader(h) + }) + } + + t.Run("from our contact", func(t *testing.T) { + require.Equal(t, 120*time.Second, grantedExpiry(withContact(*contact, "120"), contact, time.Hour)) + }) + t.Run("port defaults to 5060", func(t *testing.T) { + u := *contact + u.Port = 0 + require.Equal(t, 120*time.Second, grantedExpiry(withContact(u, "120"), contact, time.Hour)) + }) + t.Run("our binding wins over another one", func(t *testing.T) { + other := *contact + other.Host = "10.0.0.2" + resp := withContact(other, "60") + resp.AppendHeader(&sip.ContactHeader{Address: *contact, Params: sip.HeaderParams{{K: "expires", V: "120"}}}) + require.Equal(t, 120*time.Second, grantedExpiry(resp, contact, time.Hour)) + }) + t.Run("falls back to the shortest binding", func(t *testing.T) { + // Some registrars rewrite the Contact to the address they observed, so ours is not in + // the list. Refreshing early is safe; letting the binding lapse is not. + rewritten := *contact + rewritten.Host = "203.0.113.7" + rewritten.Port = 40000 + resp := withContact(rewritten, "60") + other := *contact + other.Host = "10.0.0.2" + resp.AppendHeader(&sip.ContactHeader{Address: other, Params: sip.HeaderParams{{K: "expires", V: "600"}}}) + require.Equal(t, 60*time.Second, grantedExpiry(resp, contact, time.Hour)) + }) + t.Run("from the expires header", func(t *testing.T) { + resp := newResp(func(resp *sip.Response) { + resp.AppendHeader(sip.NewHeader("Expires", "300")) + }) + require.Equal(t, 300*time.Second, grantedExpiry(resp, contact, time.Hour)) + }) + t.Run("contact wins over the expires header", func(t *testing.T) { + resp := withContact(*contact, "120") + resp.AppendHeader(sip.NewHeader("Expires", "300")) + require.Equal(t, 120*time.Second, grantedExpiry(resp, contact, time.Hour)) + }) + t.Run("falls back to what we asked for", func(t *testing.T) { + require.Equal(t, time.Hour, grantedExpiry(withContact(*contact, ""), contact, time.Hour)) + }) + t.Run("ignores a malformed value", func(t *testing.T) { + require.Equal(t, time.Hour, grantedExpiry(withContact(*contact, "soon"), contact, time.Hour)) + }) +} + +func TestRegistrantAnswersProxyChallenge(t *testing.T) { + var ( + mu sync.Mutex + chal *digest.Challenge + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if req.GetHeader("Proxy-Authorization") == nil { + c := digest.Challenge{Realm: testRegRealm, Nonce: strconv.Itoa(rand.Int()), Algorithm: "MD5"} + chal = &c + resp := sip.NewResponseFromRequest(req, sip.StatusProxyAuthRequired, "Proxy Auth Required", nil) + resp.AppendHeader(sip.NewHeader("Proxy-Authenticate", c.String())) + return resp + } + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Password: testRegPass}) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(2) + require.Nil(t, reqs[1].GetHeader("Authorization"), "a 407 must be answered with Proxy-Authorization") + h := reqs[1].GetHeader("Proxy-Authorization") + require.NotNil(t, h) + cred, err := digest.ParseCredentials(h.Value()) + require.NoError(t, err) + mu.Lock() + c := chal + mu.Unlock() + want, err := digest.Digest(c, digest.Options{ + Method: "REGISTER", URI: cred.URI, Username: testRegUser, Password: testRegPass, + }) + require.NoError(t, err) + require.Equal(t, want.Response, cred.Response) + + require.Eventually(t, func() bool { return r.State().Registered }, 5*time.Second, 10*time.Millisecond) +} + +func TestRegistrantRetriesStaleNonce(t *testing.T) { + // The refresh reuses the cached challenge and the registrar has since expired the nonce. + // A stale challenge must be answered with the new nonce, not read as a wrong password. + var ( + mu sync.Mutex + nonce = "first-nonce" + accepted int + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if h := req.GetHeader("Authorization"); h != nil { + cred, err := digest.ParseCredentials(h.Value()) + if err == nil && cred.Nonce == nonce { + accepted++ + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", "1")) + if accepted == 1 { + nonce = "second-nonce" // expire it right after the first success + } + return resp + } + } + c := digest.Challenge{Realm: testRegRealm, Nonce: nonce, Algorithm: "MD5", Stale: true} + resp := sip.NewResponseFromRequest(req, sip.StatusUnauthorized, "Unauthorized", nil) + resp.AppendHeader(sip.NewHeader("WWW-Authenticate", c.String())) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Password: testRegPass}) + r.Start() + t.Cleanup(r.Stop) + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return accepted >= 2 + }, 15*time.Second, 20*time.Millisecond, "the refresh never recovered from the stale nonce") + require.NoError(t, r.State().Error) +} + +func TestRegistrantKeepsMinExpiresAcrossRefreshes(t *testing.T) { + // The registrar refuses anything under minExpires but grants only a second, so a refresh + // follows immediately. It must ask for the negotiated value directly rather than collect + // another 423 every time. + const minExpires = 600 + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + exp, _ := strconv.Atoi(req.GetHeader("Expires").Value()) + if exp > 0 && exp < minExpires { + resp := sip.NewResponseFromRequest(req, sip.StatusIntervalToBrief, "Interval Too Brief", nil) + resp.AppendHeader(sip.NewHeader("Min-Expires", strconv.Itoa(minExpires))) + return resp + } + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", "1")) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Expiry: 60 * time.Second}) + r.Start() + t.Cleanup(r.Stop) + + reqs := reg.waitRequests(3) + require.Equal(t, "60", reqs[0].GetHeader("Expires").Value()) + require.Equal(t, "600", reqs[1].GetHeader("Expires").Value()) + require.Equal(t, "600", reqs[2].GetHeader("Expires").Value(), "the refresh must not be refused again") +} + +func TestRegistrantKeepsPingingAfterRefreshFails(t *testing.T) { + // The NAT mapping the provider is already sending calls to has to be held open while + // re-registration retries, not abandoned at the first failure. + var ( + mu sync.Mutex + refused bool + gotAfter int + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + switch req.Method { + case sip.OPTIONS: + if refused { + gotAfter++ + } + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + default: + if !refused { + refused = true + // Long enough that the binding is still live when the refresh fails, short + // enough that the refresh is due almost immediately. + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", "11")) + return resp + } + return sip.NewResponseFromRequest(req, sip.StatusServiceUnavailable, "Service Unavailable", nil) + } + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{ + Keepalive: 20 * time.Millisecond, + }) + r.Start() + t.Cleanup(r.Stop) + + require.Eventually(t, func() bool { + return r.State().Error != nil + }, 5*time.Second, 10*time.Millisecond, "the refresh never failed") + // The binding has not expired yet, so it is still reported and still held open. + require.True(t, r.State().Registered) + + mu.Lock() + before := gotAfter + mu.Unlock() + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return gotAfter > before + }, 5*time.Second, 10*time.Millisecond, "keepalives stopped once the refresh failed") +} + +func TestRegistrantAuthenticatesUnregister(t *testing.T) { + // Some registrars challenge the un-REGISTER with a fresh nonce; leaving the binding behind + // would send inbound calls to a node that has stopped. + var ( + mu sync.Mutex + nonce = "first-nonce" + final *sip.Request + ) + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if h := req.GetHeader("Authorization"); h != nil { + if cred, err := digest.ParseCredentials(h.Value()); err == nil && cred.Nonce == nonce { + if req.GetHeader("Expires").Value() == "0" { + final = req + } else { + // Expire the nonce, so the un-REGISTER's cached credentials are refused + // and it has to answer a fresh challenge. + nonce = "unregister-nonce" + } + return sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + } + } + c := digest.Challenge{Realm: testRegRealm, Nonce: nonce, Algorithm: "MD5"} + resp := sip.NewResponseFromRequest(req, sip.StatusUnauthorized, "Unauthorized", nil) + resp.AppendHeader(sip.NewHeader("WWW-Authenticate", c.String())) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{Password: testRegPass}) + r.Start() + require.Eventually(t, func() bool { return r.State().Registered }, 5*time.Second, 10*time.Millisecond) + + r.Stop() + mu.Lock() + last := final + mu.Unlock() + require.NotNil(t, last, "the un-REGISTER was never accepted") + require.NotNil(t, last.Contact(), "the un-REGISTER must name our own binding, not all of them") + require.Len(t, reg.received(), 4, "expected a challenge and a retry for both the REGISTER and the un-REGISTER") +} + +func TestWaitForSignalingSocket(t *testing.T) { + log := slog.New(logger.ToSlogHandler(logger.LogRLogger(logr.Discard()))) + localIP, err := config.GetLocalIP() + require.NoError(t, err) + ua, err := sipgo.NewUA(sipgo.WithUserAgent(UserAgent), sipgo.WithUserAgentLogger(log)) + require.NoError(t, err) + t.Cleanup(func() { _ = ua.Close() }) + + // Nothing is serving yet, so the wait has to time out rather than report a socket that + // would send registrations from an ephemeral port. + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + require.ErrorIs(t, waitForSignalingSocket(ctx, ua.TransportLayer(), "192.0.2.1:5060"), context.DeadlineExceeded) + + srv, err := sipgo.NewServer(ua, sipgo.WithServerLogger(log)) + require.NoError(t, err) + lis, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IP(localIP.AsSlice()), Port: 0}) + require.NoError(t, err) + go func() { + _ = srv.ServeUDP(lis) + }() + t.Cleanup(func() { + _ = srv.Close() + _ = lis.Close() + }) + + ctx, cancel = context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + require.NoError(t, waitForSignalingSocket(ctx, ua.TransportLayer(), "192.0.2.1:5060")) +} + +func TestValidateRegistrations(t *testing.T) { + t.Run("bad registrar", func(t *testing.T) { + require.Error(t, validateRegistrations(&config.Config{ + SIPRegistrations: []config.SIPRegistrationConfig{{Registrar: "sip:", Username: "alice"}}, + })) + }) + t.Run("tls registrar without tls config", func(t *testing.T) { + require.Error(t, validateRegistrations(&config.Config{ + SIPRegistrations: []config.SIPRegistrationConfig{{Registrar: "sips:sip.example.com", Username: "alice"}}, + })) + require.NoError(t, validateRegistrations(&config.Config{ + TLS: &config.TLSConfig{}, + SIPRegistrations: []config.SIPRegistrationConfig{{Registrar: "sips:sip.example.com", Username: "alice"}}, + })) + }) + t.Run("ok", func(t *testing.T) { + require.NoError(t, validateRegistrations(&config.Config{ + SIPRegistrations: []config.SIPRegistrationConfig{{Registrar: "sip.example.com", Username: "alice"}}, + })) + }) +} + +func TestRegistrantStillBound(t *testing.T) { + r, _ := newTestRegistrant(t, freeAddr(t), config.SIPRegistrationConfig{}) + require.False(t, r.stillBound(), "nothing has been registered yet") + + r.bound, r.boundUntil = true, time.Now().Add(time.Minute) + require.True(t, r.stillBound()) + + r.boundUntil = time.Now().Add(-time.Second) + require.False(t, r.stillBound(), "a lapsed binding is not ours to claim") +} + +func TestRegistrantReportsLapsedBinding(t *testing.T) { + // The registrar grants a short lifetime and then stops accepting refreshes. The binding + // survives the first failure, then lapses, and must stop being reported at that point: + // this is what the registrations_active gauge is read for. + var mu sync.Mutex + accepted := false + reg := newTestRegistrar(t, func(req *sip.Request) *sip.Response { + mu.Lock() + defer mu.Unlock() + if accepted { + return sip.NewResponseFromRequest(req, sip.StatusServiceUnavailable, "Service Unavailable", nil) + } + accepted = true + resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) + resp.AppendHeader(sip.NewHeader("Expires", "3")) + return resp + }) + + r, _ := newTestRegistrant(t, reg.addr, config.SIPRegistrationConfig{}) + r.Start() + t.Cleanup(r.Stop) + + // The refresh is due at 1.5s and fails, but the binding is good until 3s. + require.Eventually(t, func() bool { + st := r.State() + return st.Error != nil && st.Registered + }, 3*time.Second, 10*time.Millisecond, "the failed refresh should not drop a live binding") + + require.Eventually(t, func() bool { + return !r.State().Registered + }, 10*time.Second, 20*time.Millisecond, "the lapsed binding was still reported as live") + require.Error(t, r.State().Error) +} diff --git a/pkg/sip/service.go b/pkg/sip/service.go index 041433c93..64358f625 100644 --- a/pkg/sip/service.go +++ b/pkg/sip/service.go @@ -69,6 +69,8 @@ type Service struct { srv *Server closers []io.Closer + registrants []*registrant + mu sync.Mutex pendingTransfers map[LocalTag]*PendingTransfer } @@ -107,6 +109,9 @@ func NewService(region string, conf *config.Config, mon *stats.Monitor, log logg if err != nil { return nil, err } + if err := validateRegistrations(conf); err != nil { + return nil, err + } const placeholder = "${IP}" if strings.Contains(s.conf.SIPHostname, placeholder) { @@ -191,7 +196,19 @@ func (s *Service) ActiveCalls() ActiveCalls { return st } +// StopRegistrations withdraws this node's registrations. Calling it before draining stops a +// provider from routing new inbound calls to a node that is shutting down; Stop calls it too, +// so it is safe either way. +func (s *Service) StopRegistrations() { + for _, r := range s.registrants { + r.Stop() + } + s.registrants = nil +} + func (s *Service) Stop() { + // Unregister while the SIP client is still usable. + s.StopRegistrations() s.cli.Stop() s.srv.Stop() s.mon.Stop() @@ -311,10 +328,67 @@ func (s *Service) Start() error { if err := s.srv.Start(ua, s.sconf, tlsConf, s.cli.OnRequest); err != nil { return err } + if err := s.startRegistrations(ua); err != nil { + return err + } s.log.Debugw("sip service ready") return nil } +// startRegistrations registers this node with the configured registrars. It runs after the +// server is listening so that registrations go out from the SIP signaling socket. +func (s *Service) startRegistrations(ua *sipgo.UserAgent) error { + if len(s.conf.SIPRegistrations) == 0 { + return nil + } + var registrants []*registrant + for _, rc := range s.conf.SIPRegistrations { + r, err := newRegistrant(s.log, s.cli.sipCli, s.mon, s.conf, s.sconf, rc) + if err != nil { + return err + } + registrants = append(registrants, r) + } + // Only UDP reuses the server's listener, and one wait covers every registration on it. + // Registering before it exists would send from an ephemeral port for good, so this is + // fatal rather than a warning. + if tp := ua.TransportLayer(); tp != nil { + for _, r := range registrants { + if r.transport != strings.ToUpper(string(TransportUDP)) { + continue + } + ctx, cancel := context.WithTimeout(context.Background(), signalingSocketTimeout) + err := waitForSignalingSocket(ctx, tp, r.dest()) + cancel() + if err != nil { + return fmt.Errorf("SIP signaling socket did not come up: %w", err) + } + break + } + } + s.registrants = registrants + s.log.Infow("registering with SIP registrars", "count", len(registrants)) + for _, r := range registrants { + r.Start() + } + return nil +} + +// validateRegistrations reports configuration this node cannot act on, at construction time +// rather than once the listeners are already up. +func validateRegistrations(conf *config.Config) error { + for i, rc := range conf.SIPRegistrations { + u, err := parseRegistrarURI(rc.Registrar) + if err != nil { + return fmt.Errorf("sip_registrations[%d]: %w", i, err) + } + if registrarTransport(&u) == TransportTLS && conf.TLS == nil { + return fmt.Errorf("sip_registrations[%d]: a TLS registrar requires the tls config", i) + } + } + return nil +} + func (s *Service) CreateSIPParticipant(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) (*rpc.InternalCreateSIPParticipantResponse, error) { resp, err := s.cli.CreateSIPParticipant(ctx, req) return resp, siperrors.ApplySIPStatus(err) diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go index 47679ffe3..ad8ca8b4d 100644 --- a/pkg/stats/monitor.go +++ b/pkg/stats/monitor.go @@ -80,6 +80,7 @@ type Monitor struct { transfersSucceeded *prometheus.CounterVec transfersFailed *prometheus.CounterVec transfersActive *prometheus.GaugeVec + registrationsActive *prometheus.GaugeVec cpu *hwstats.CPUStats maxUtilization float64 @@ -164,6 +165,14 @@ func (m *Monitor) Start(conf *config.Config) error { ConstLabels: prometheus.Labels{"node_id": conf.NodeID}, }, []string{"dir", "to"})) + m.registrationsActive = mustRegister(m, prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "livekit", + Subsystem: "sip", + Name: "registrations_active", + Help: "Whether this node currently has a binding at a SIP registrar", + ConstLabels: prometheus.Labels{"node_id": conf.NodeID}, + }, []string{"registrar"})) + m.callsTerminated = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "livekit", Subsystem: "sip", @@ -496,6 +505,16 @@ func (c *CallMonitor) SDPSize(sz int, isOffer bool) { c.m.sdpSize.WithLabelValues(typ).Observe(float64(sz)) } +// RegistrationActive records whether a SIP registration currently has a binding. Inbound calls +// on a registered trunk stop arriving when it does not, so it is worth alerting on. +func (m *Monitor) RegistrationActive(registrar string, active bool) { + v := float64(0) + if active { + v = 1 + } + m.registrationsActive.WithLabelValues(registrar).Set(v) +} + func (m *Monitor) TransferStarted(dir CallDir) { m.transfersTotal.WithLabelValues(dir.String()).Inc() m.transfersActive.WithLabelValues(dir.String()).Inc() diff --git a/test/cloud/service.go b/test/cloud/service.go index 4d83d6a07..210c85e99 100644 --- a/test/cloud/service.go +++ b/test/cloud/service.go @@ -18,11 +18,14 @@ func NewService(conf *IntegrationConfig, bus psrpc.MessageBus) (*service.Service return nil, err } - sipsrv, err := sip.NewService("", conf.Config, mon, logger.GetLogger(), func(projectID string, _ *rpc.SIPCallObservability, _ *livekit.SIPCallInfo) sip.StateHandler { return sip.NewRPCStateHandler(psrpcClient) }) + sipsrv, err := sip.NewService("", conf.Config, mon, logger.GetLogger(), func(projectID string, _ *rpc.SIPCallObservability, _ *livekit.SIPCallInfo) sip.StateHandler { + return sip.NewRPCStateHandler(psrpcClient) + }) if err != nil { return nil, err } svc := service.NewService(conf.Config, logger.GetLogger(), sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon) + svc.SetSIPServiceDrain(sipsrv.StopRegistrations) sipsrv.SetHandler(svc) if err = sipsrv.Start(); err != nil { From 4eaf04010e3ef9a0f7167051d728301b46904508 Mon Sep 17 00:00:00 2001 From: Artur Schaback Date: Tue, 4 Aug 2026 04:06:14 +0300 Subject: [PATCH 2/3] fix: address review of the SIP registration change Label the registrations_active gauge with the address-of-record as well as the registrar, so several accounts on one provider do not share a series and mask each other. Never schedule a refresh sooner than half the granted lifetime. Subtracting a fixed margin alone made a lifetime just over that margin refresh far more often than a slightly shorter one. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018vEyHdjPUV3BrxxN2m41pT --- pkg/sip/register.go | 8 +++++--- pkg/sip/register_test.go | 15 ++++++++++----- pkg/stats/monitor.go | 9 +++++---- 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/pkg/sip/register.go b/pkg/sip/register.go index 04d6d9c8e..b1108c531 100644 --- a/pkg/sip/register.go +++ b/pkg/sip/register.go @@ -258,7 +258,7 @@ func (r *registrant) State() RegistrationState { func (r *registrant) setState(st RegistrationState) { r.state.Store(&st) if r.mon != nil { - r.mon.RegistrationActive(r.registrar.String(), st.Registered) + r.mon.RegistrationActive(r.registrar.String(), r.aor.String(), st.Registered) } } @@ -371,9 +371,11 @@ func refreshAfter(d time.Duration) time.Duration { } else if margin > registerMaxRefreshMargin { margin = registerMaxRefreshMargin } + // Never wait less than half the lifetime: subtracting a fixed margin alone would make a + // grant just over it refresh far more often than a slightly shorter one. after := d - margin - if d <= margin { - after = d / 2 + if half := d / 2; after < half { + after = half } if after < registerMinRefreshInterval { after = registerMinRefreshInterval diff --git a/pkg/sip/register_test.go b/pkg/sip/register_test.go index 724615966..aa7817476 100644 --- a/pkg/sip/register_test.go +++ b/pkg/sip/register_test.go @@ -553,15 +553,20 @@ func TestRefreshAfter(t *testing.T) { }{ {granted: time.Second, want: registerMinRefreshInterval}, // interval floor {granted: 10 * time.Second, want: 5 * time.Second}, // margin floor + {granted: 11 * time.Second, want: 5500 * time.Millisecond}, {granted: 60 * time.Second, want: 50 * time.Second}, {granted: 600 * time.Second, want: 540 * time.Second}, {granted: time.Hour, want: time.Hour - time.Minute}, // margin ceiling } + var prev time.Duration for _, c := range cases { t.Run(c.granted.String(), func(t *testing.T) { got := refreshAfter(c.granted) require.Equal(t, c.want, got) require.LessOrEqual(t, got, c.granted, "a refresh must not be scheduled past the expiry") + // A longer lifetime must never mean a shorter interval. + require.GreaterOrEqual(t, got, prev) + prev = got }) } } @@ -761,10 +766,10 @@ func TestRegistrantKeepsPingingAfterRefreshFails(t *testing.T) { default: if !refused { refused = true - // Long enough that the binding is still live when the refresh fails, short - // enough that the refresh is due almost immediately. + // The refresh comes due at half of this, leaving the binding live for the + // same span again while it fails. resp := sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil) - resp.AppendHeader(sip.NewHeader("Expires", "11")) + resp.AppendHeader(sip.NewHeader("Expires", "10")) return resp } return sip.NewResponseFromRequest(req, sip.StatusServiceUnavailable, "Service Unavailable", nil) @@ -779,7 +784,7 @@ func TestRegistrantKeepsPingingAfterRefreshFails(t *testing.T) { require.Eventually(t, func() bool { return r.State().Error != nil - }, 5*time.Second, 10*time.Millisecond, "the refresh never failed") + }, 8*time.Second, 10*time.Millisecond, "the refresh never failed") // The binding has not expired yet, so it is still reported and still held open. require.True(t, r.State().Registered) @@ -790,7 +795,7 @@ func TestRegistrantKeepsPingingAfterRefreshFails(t *testing.T) { mu.Lock() defer mu.Unlock() return gotAfter > before - }, 5*time.Second, 10*time.Millisecond, "keepalives stopped once the refresh failed") + }, 4*time.Second, 10*time.Millisecond, "keepalives stopped once the refresh failed") } func TestRegistrantAuthenticatesUnregister(t *testing.T) { diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go index ad8ca8b4d..4d5f271f6 100644 --- a/pkg/stats/monitor.go +++ b/pkg/stats/monitor.go @@ -171,7 +171,7 @@ func (m *Monitor) Start(conf *config.Config) error { Name: "registrations_active", Help: "Whether this node currently has a binding at a SIP registrar", ConstLabels: prometheus.Labels{"node_id": conf.NodeID}, - }, []string{"registrar"})) + }, []string{"registrar", "aor"})) m.callsTerminated = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "livekit", @@ -506,13 +506,14 @@ func (c *CallMonitor) SDPSize(sz int, isOffer bool) { } // RegistrationActive records whether a SIP registration currently has a binding. Inbound calls -// on a registered trunk stop arriving when it does not, so it is worth alerting on. -func (m *Monitor) RegistrationActive(registrar string, active bool) { +// on a registered trunk stop arriving when it does not, so it is worth alerting on. The +// address-of-record is a label of its own because several accounts can share one registrar. +func (m *Monitor) RegistrationActive(registrar, aor string, active bool) { v := float64(0) if active { v = 1 } - m.registrationsActive.WithLabelValues(registrar).Set(v) + m.registrationsActive.WithLabelValues(registrar, aor).Set(v) } func (m *Monitor) TransferStarted(dir CallDir) { From 65fff823f6471aa671fedebdcd4d200099b13c6c Mon Sep 17 00:00:00 2001 From: Artur Schaback Date: Tue, 4 Aug 2026 04:15:35 +0300 Subject: [PATCH 3/3] fix: keep the registrar's challenge out of logs A challenge that fails to parse no longer has its header value repeated in the error. It carries the registrar's nonce and opaque, and the retry loop writes that error to the log on every attempt. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018vEyHdjPUV3BrxxN2m41pT --- pkg/sip/register.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/sip/register.go b/pkg/sip/register.go index b1108c531..24b303883 100644 --- a/pkg/sip/register.go +++ b/pkg/sip/register.go @@ -539,7 +539,9 @@ func (r *registrant) setChallenge(resp *sip.Response) error { } chal, err := digest.ParseChallenge(h.Value()) if err != nil { - return fmt.Errorf("invalid %s challenge %q: %w", name, h.Value(), err) + // The header itself is not repeated: it carries the registrar's nonce and opaque, + // which do not belong in logs that are kept. + return fmt.Errorf("invalid %s challenge: %w", name, err) } // A repeat of a nonce we already answered, without stale=true, means the credentials were // wrong rather than expired. Retrying would send the same digest and loop.