diff --git a/README.md b/README.md index faed8dfa..c3c98605 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 7f3bd27a..f16e08ab 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 8192660d..b12706db 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 00000000..bd5a1c92 --- /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 dbf61253..3011c409 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 00000000..24b30388 --- /dev/null +++ b/pkg/sip/register.go @@ -0,0 +1,714 @@ +// 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(), r.aor.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 + } + // 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 half := d / 2; after < half { + after = half + } + 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 { + // 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. + 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 00000000..aa781747 --- /dev/null +++ b/pkg/sip/register_test.go @@ -0,0 +1,939 @@ +// 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: 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 + }) + } +} + +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 + // 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", "10")) + 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 + }, 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) + + mu.Lock() + before := gotAfter + mu.Unlock() + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return gotAfter > before + }, 4*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 041433c9..64358f62 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 47679ffe..4d5f271f 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", "aor"})) + m.callsTerminated = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "livekit", Subsystem: "sip", @@ -496,6 +505,17 @@ 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. 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, aor).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 4d83d6a0..210c85e9 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 {