diff --git a/pkg/sip/client.go b/pkg/sip/client.go index fa64bebdb..3d2ee52d7 100644 --- a/pkg/sip/client.go +++ b/pkg/sip/client.go @@ -50,6 +50,9 @@ import ( type SIPClient interface { TransactionRequest(req *sip.Request, options ...sipgo.ClientRequestOption) (sip.ClientTransaction, error) WriteRequest(req *sip.Request, options ...sipgo.ClientRequestOption) error + // ResolveTargets returns the addresses to try for a request URI, in order. + // port is the port from the URI, or 0 when it carries none. + ResolveTargets(ctx context.Context, network, host string, port int, sipScheme string) ([]netip.AddrPort, error) Close() error } diff --git a/pkg/sip/outbound.go b/pkg/sip/outbound.go index 89ef94b19..968fbe066 100644 --- a/pkg/sip/outbound.go +++ b/pkg/sip/outbound.go @@ -987,12 +987,10 @@ func (c *sipOutbound) Invite(ctx context.Context, user, pass string, headers map defer c.mu.Unlock() var ( - sipHeaders Headers - authHeader = "" - authHeaderRespName string - req *sip.Request - resp *sip.Response - err error + sipHeaders Headers + req *sip.Request + resp *sip.Response + err error ) if keys := maps.Keys(headers); len(keys) != 0 { sort.Strings(keys) @@ -1000,14 +998,118 @@ func (c *sipOutbound) Invite(ctx context.Context, user, pass string, headers map sipHeaders = append(sipHeaders, sip.NewHeader(key, headers[key])) } } + + dests := c.inviteDestinations(ctx, headers) + for di, dest := range dests { + req, resp, err = c.inviteDest(ctx, dest, user, pass, sdpOffer, sipHeaders, setState) + if err == nil { + break + } + var next errTryNextDest + if !errors.As(err, &next) { + return nil, err + } + // Out of addresses, or the call is going away anyway. Report the + // underlying failure, not the fact that we considered a retry. + if di == len(dests)-1 || ctx.Err() != nil { + return nil, next.err + } + c.log.Infow("INVITE failed, trying next destination", + "failed", dest, "next", dests[di+1], "error", next.err) + } + if err != nil { + return nil, err + } + + c.invite, c.inviteOk = req, resp + toHeader := resp.To() + if toHeader == nil { + return nil, psrpc.NewErrorf(psrpc.Internal, "no To header in INVITE response") + } + var ok bool + c.tag, ok = getTagFrom(toHeader.Params) + if !ok { + return nil, psrpc.NewErrorf(psrpc.Internal, "no tag in To header in INVITE response") + } + + applyInviteResponse(req, resp) + return c.inviteOk.Body(), nil +} + +// errTryNextDest marks a failure as specific to the destination we tried, so the +// caller can move on to the next resolved address. Statuses that answer the +// request itself (busy, not found, carrier blocks) are not wrapped: another +// server would give the same answer. +type errTryNextDest struct{ err error } + +func (e errTryNextDest) Error() string { return e.err.Error() } +func (e errTryNextDest) Unwrap() error { return e.err } + +// maxInviteDests bounds how many addresses we try before giving up, so call +// setup stays bounded when a carrier publishes a long list. +const maxInviteDests = 3 + +// inviteDestinations resolves the request URI to the addresses to try, in order. +// A single empty destination means "let the transport layer resolve it", which +// is the behaviour when we can't resolve or when a Route header is in play. +// +// Resolution lives in sipgo, which owns the RFC 3263 rules. All this does is +// decide how many of the results are worth trying. +func (c *sipOutbound) inviteDestinations(ctx context.Context, headers map[string]string) []string { + unresolved := []string{""} + + // With a Route header the request goes to the proxy, not to the request URI, + // so resolving the request URI here would send it to the wrong host. + if len(c.routeHeaders) != 0 { + return unresolved + } + for k := range headers { + if strings.EqualFold(k, "Route") { + return unresolved + } + } + + targets, err := c.c.sipCli.ResolveTargets(ctx, uriTransport(c.uri), c.uri.Host, c.uri.Port, c.uri.Scheme) + if err != nil { + c.log.Warnw("could not resolve destination, falling back to transport resolution", err, + "host", c.uri.Host, "port", c.uri.Port) + return unresolved + } + if len(targets) > maxInviteDests { + c.log.Infow("more destinations than we will try", + "host", c.uri.Host, "resolved", len(targets), "trying", maxInviteDests) + targets = targets[:maxInviteDests] + } + dests := make([]string, 0, len(targets)) + for _, t := range targets { + dests = append(dests, t.String()) + } + if len(dests) == 0 { + return unresolved + } + return dests +} + +// inviteDest runs the INVITE and its authentication retries against a single +// destination. Pinning the destination for the whole exchange keeps the +// challenge and the authenticated INVITE on the same server. +func (c *sipOutbound) inviteDest(ctx context.Context, dest, user, pass string, sdpOffer []byte, sipHeaders Headers, setState sipRespFunc) (*sip.Request, *sip.Response, error) { + var ( + authHeader = "" + authHeaderRespName string + req *sip.Request + resp *sip.Response + err error + ) authLoop: for try := 0; ; try++ { if try >= 5 { - return nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthMaxRetry) + return nil, nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthMaxRetry) } - req, resp, err = c.attemptInvite(ctx, sip.CallIDHeader(c.callID), sdpOffer, authHeaderRespName, authHeader, sipHeaders, setState) + req, resp, err = c.attemptInvite(ctx, dest, sip.CallIDHeader(c.callID), sdpOffer, authHeaderRespName, authHeader, sipHeaders, setState) if err != nil { - return nil, err + // We never reached this server, or it never answered. + return nil, nil, errTryNextDest{err} } var authHeaderName string switch resp.StatusCode { @@ -1019,9 +1121,19 @@ authLoop: Status: resp.Reason, } if blocked := carrierBlockFromResponse(resp, st); blocked != nil { - return nil, fmt.Errorf("INVITE blocked by carrier: %w", blocked) + return nil, nil, fmt.Errorf("INVITE blocked by carrier: %w", blocked) + } + err = fmt.Errorf("unexpected status from INVITE response: %w", st) + if resp.StatusCode >= 500 && resp.StatusCode < 600 { + // The server failed rather than answered. Another one may not. + return nil, nil, errTryNextDest{err} } - return nil, fmt.Errorf("unexpected status from INVITE response: %w", st) + return nil, nil, err + case sip.StatusRequestTimeout: + return nil, nil, errTryNextDest{fmt.Errorf("INVITE timed out: %w", &livekit.SIPStatus{ + Code: livekit.SIPStatusCode(resp.StatusCode), + Status: resp.Reason, + })} case sip.StatusBadRequest, sip.StatusNotFound, sip.StatusTemporarilyUnavailable, @@ -1037,9 +1149,9 @@ authLoop: st.Status = s.Value() } if blocked := carrierBlockFromResponse(resp, st); blocked != nil { - return nil, fmt.Errorf("INVITE blocked by carrier: %w", blocked) + return nil, nil, fmt.Errorf("INVITE blocked by carrier: %w", blocked) } - return nil, fmt.Errorf("INVITE failed: %w", st) + return nil, nil, fmt.Errorf("INVITE failed: %w", st) case sip.StatusUnauthorized: authHeaderName = "WWW-Authenticate" authHeaderRespName = "Authorization" @@ -1050,20 +1162,20 @@ authLoop: c.log.Infow("auth requested", "status", resp.StatusCode, "body", string(resp.Body())) // auth required if user == "" || pass == "" { - return nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthMissingCreds) + return nil, nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthMissingCreds) } headerVal := resp.GetHeader(authHeaderName) if headerVal == nil { - return nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthNoHeader) + return nil, nil, psrpc.NewError(psrpc.FailedPrecondition, ErrAuthNoHeader) } challengeStr := headerVal.Value() challenge, err := digest.ParseChallenge(challengeStr) if err != nil { - return nil, psrpc.NewErrorf(psrpc.Internal, "invalid challenge %q: %v", challengeStr, err) + return nil, nil, psrpc.NewErrorf(psrpc.Internal, "invalid challenge %q: %v", challengeStr, err) } toHeader := resp.To() if toHeader == nil { - return nil, psrpc.NewErrorf(psrpc.Internal, "no 'To' header on Response") + return nil, nil, psrpc.NewErrorf(psrpc.Internal, "no 'To' header on Response") } cred, err := digest.Digest(challenge, digest.Options{ @@ -1073,25 +1185,26 @@ authLoop: Password: pass, }) if err != nil { - return nil, err + return nil, nil, err } authHeader = cred.String() // Try again with a computed digest } + return req, resp, nil +} - c.invite, c.inviteOk = req, resp - toHeader := resp.To() - if toHeader == nil { - return nil, psrpc.NewErrorf(psrpc.Internal, "no To header in INVITE response") +// uriTransport returns the transport a request to uri will use, matching how +// sipgo derives it from the request. +func uriTransport(uri *sip.Uri) string { + if uri.UriParams != nil { + if v, ok := uri.UriParams.Get("transport"); ok && v != "" { + return strings.ToLower(v) + } } - var ok bool - c.tag, ok = getTagFrom(toHeader.Params) - if !ok { - return nil, psrpc.NewErrorf(psrpc.Internal, "no tag in To header in INVITE response") + if strings.EqualFold(uri.Scheme, "sips") { + return "tls" } - - applyInviteResponse(req, resp) - return c.inviteOk.Body(), nil + return "udp" } // applyInviteResponse rewrites the INVITE request in place so ACK/BYE built from @@ -1146,10 +1259,15 @@ func (c *sipOutbound) AckInviteOK(ctx context.Context) error { return c.c.sipCli.WriteRequest(sip.NewAckRequest(c.invite, c.inviteOk, nil)) } -func (c *sipOutbound) attemptInvite(ctx context.Context, callID sip.CallIDHeader, offer []byte, authHeaderName, authHeader string, headers Headers, setState sipRespFunc) (*sip.Request, *sip.Response, error) { +func (c *sipOutbound) attemptInvite(ctx context.Context, dest string, callID sip.CallIDHeader, offer []byte, authHeaderName, authHeader string, headers Headers, setState sipRespFunc) (*sip.Request, *sip.Response, error) { ctx, span := Tracer.Start(ctx, "sip.outbound.attemptInvite") defer span.End() req := sip.NewRequest(sip.INVITE, *c.uri) + if dest != "" { + // Send to the address we resolved, keeping the request URI as configured. + // The port belongs to this address; it must not come from anywhere else. + req.SetDestination(dest) + } c.setCSeq(req) req.RemoveHeader("Call-ID") req.AppendHeader(&callID) diff --git a/pkg/sip/outbound_test.go b/pkg/sip/outbound_test.go index e01a1953f..fa4f0f962 100644 --- a/pkg/sip/outbound_test.go +++ b/pkg/sip/outbound_test.go @@ -16,7 +16,13 @@ package sip import ( "context" + "encoding/binary" + "errors" "fmt" + "net" + "net/netip" + "strings" + "sync" "testing" "time" @@ -44,6 +50,10 @@ func (c *recordingSIPClient) WriteRequest(req *sip.Request, _ ...sipgo.ClientReq return nil } +func (c *recordingSIPClient) ResolveTargets(_ context.Context, _, _ string, _ int, _ string) ([]netip.AddrPort, error) { + return nil, errors.New("not resolved in this test") +} + func (c *recordingSIPClient) Close() error { return nil } func (c *recordingSIPClient) methods() []sip.RequestMethod { @@ -711,3 +721,294 @@ func TestBuildOutboundHeaders(t *testing.T) { } }) } + +// fakeDNS is a minimal UDP DNS server answering A and SRV queries from a table, +// so outbound resolution runs through sipgo for real in tests. +type fakeDNS struct { + conn *net.UDPConn + a map[string][]string + srv map[string][]fakeSRV + + mu sync.Mutex + seen map[string]int + + done sync.WaitGroup +} + +type fakeSRV struct { + target string + port uint16 +} + +const ( + dnsTypeA = 1 + dnsTypeSRV = 33 +) + +func newFakeDNS(t *testing.T, a map[string][]string, srv map[string][]fakeSRV) *fakeDNS { + t.Helper() + conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + require.NoError(t, err) + d := &fakeDNS{conn: conn, a: a, srv: srv, seen: map[string]int{}} + d.done.Add(1) + go d.serve() + // Closing the socket makes the read in serve fail, which ends the goroutine. + // Wait for it so it cannot outlive the test. + t.Cleanup(func() { + _ = conn.Close() + d.done.Wait() + }) + return d +} + +func (d *fakeDNS) resolver() *net.Resolver { + addr := d.conn.LocalAddr().String() + return &net.Resolver{ + PreferGo: true, + Dial: func(ctx context.Context, _, _ string) (net.Conn, error) { + var dialer net.Dialer + return dialer.DialContext(ctx, "udp", addr) + }, + } +} + +func (d *fakeDNS) queries(name string, qtype uint16) int { + d.mu.Lock() + defer d.mu.Unlock() + if !strings.HasSuffix(name, ".") { + name += "." + } + return d.seen[strings.ToLower(name)+"/"+string(rune(qtype))] +} + +func (d *fakeDNS) serve() { + defer d.done.Done() + buf := make([]byte, 512) + for { + n, from, err := d.conn.ReadFromUDP(buf) + if err != nil { + return + } + if resp := d.respond(buf[:n]); resp != nil { + _, _ = d.conn.WriteToUDP(resp, from) + } + } +} + +func (d *fakeDNS) respond(q []byte) []byte { + if len(q) < 12 { + return nil + } + name, off, ok := dnsReadName(q, 12) + if !ok || off+4 > len(q) { + return nil + } + qtype := binary.BigEndian.Uint16(q[off : off+2]) + d.mu.Lock() + d.seen[strings.ToLower(name)+"/"+string(rune(qtype))]++ + d.mu.Unlock() + + var answers []byte + var count int + switch qtype { + case dnsTypeA: + for _, ip := range d.a[strings.TrimSuffix(name, ".")] { + answers = append(answers, dnsEncodeName(name)...) + answers = append(answers, dnsRRHeader(dnsTypeA, 4)...) + answers = append(answers, net.ParseIP(ip).To4()...) + count++ + } + case dnsTypeSRV: + for _, rec := range d.srv[strings.TrimSuffix(name, ".")] { + target := dnsEncodeName(rec.target) + rdata := make([]byte, 6, 6+len(target)) + binary.BigEndian.PutUint16(rdata[0:], 5) + binary.BigEndian.PutUint16(rdata[2:], 50) + binary.BigEndian.PutUint16(rdata[4:], rec.port) + rdata = append(rdata, target...) + answers = append(answers, dnsEncodeName(name)...) + answers = append(answers, dnsRRHeader(dnsTypeSRV, uint16(len(rdata)))...) + answers = append(answers, rdata...) + count++ + } + } + + resp := make([]byte, 12) + copy(resp, q[:2]) + binary.BigEndian.PutUint16(resp[2:], 0x8180) + binary.BigEndian.PutUint16(resp[4:], 1) + binary.BigEndian.PutUint16(resp[6:], uint16(count)) + if count == 0 { + binary.BigEndian.PutUint16(resp[2:], 0x8183) + } + resp = append(resp, q[12:off+4]...) + return append(resp, answers...) +} + +func dnsRRHeader(qtype, rdlen uint16) []byte { + b := make([]byte, 10) + binary.BigEndian.PutUint16(b[0:], qtype) + binary.BigEndian.PutUint16(b[2:], 1) + binary.BigEndian.PutUint32(b[4:], 60) + binary.BigEndian.PutUint16(b[8:], rdlen) + return b +} + +func dnsEncodeName(name string) []byte { + var b []byte + for _, label := range strings.Split(strings.TrimSuffix(name, "."), ".") { + b = append(b, byte(len(label))) + b = append(b, label...) + } + return append(b, 0) +} + +func dnsReadName(msg []byte, off int) (string, int, bool) { + var sb strings.Builder + for off < len(msg) { + n := int(msg[off]) + off++ + if n == 0 { + return sb.String(), off, true + } + if n > 63 || off+n > len(msg) { + return "", 0, false + } + sb.Write(msg[off : off+n]) + sb.WriteByte('.') + off += n + } + return "", 0, false +} + +const ( + testSBC1IP = "198.51.100.11" + testSBC2IP = "198.51.100.12" + testSBC1 = "sbc1.example.net" + testSBC2 = "sbc2.example.net" +) + +// twoSBCDNS reproduces a carrier layout where two servers sit on different ports +// behind one hostname whose A record lists both of their addresses. +func twoSBCDNS(t *testing.T) *fakeDNS { + return newFakeDNS(t, + map[string][]string{ + testSBC1: {testSBC1IP}, + testSBC2: {testSBC2IP}, + testInviteTargetHost: {testSBC1IP, testSBC2IP}, + }, + map[string][]fakeSRV{ + "_sip._udp." + testInviteTargetHost: { + {target: testSBC1, port: 5006}, + {target: testSBC2, port: 5008}, + }, + }, + ) +} + +func startOutboundCall(t *testing.T, cfg TestClientConfig, req *rpc.InternalCreateSIPParticipantRequest) *testSIPClient { + t.Helper() + client := NewOutboundTestClient(t, cfg) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + go func() { + _, _ = client.CreateSIPParticipant(ctx, req) + }() + + select { + case sipClient := <-createdClients: + t.Cleanup(func() { _ = sipClient.Close() }) + return sipClient + case <-time.After(2 * time.Second): + require.Fail(t, "expected test SIP client to be created") + return nil + } +} + +func nextTransaction(t *testing.T, sipClient *testSIPClient) *transactionRequest { + t.Helper() + select { + case tr := <-sipClient.transactions: + t.Cleanup(func() { tr.transaction.Terminate() }) + return tr + case <-time.After(2 * time.Second): + require.Fail(t, "expected INVITE transaction") + return nil + } +} + +// End to end: sipgo pairs each SRV target with its own port, and a 500 from one +// server moves the call to the next instead of failing it. +func TestOutboundSRVDestinationsAndFailover(t *testing.T) { + d := twoSBCDNS(t) + sipClient := startOutboundCall(t, TestClientConfig{DNSResolver: d.resolver()}, MinimalCreateSIPParticipantRequest()) + + valid := map[string]string{ + testSBC1IP + ":5006": testSBC2IP + ":5008", + testSBC2IP + ":5008": testSBC1IP + ":5006", + } + + first := nextTransaction(t, sipClient) + require.Equal(t, sip.INVITE, first.req.Method) + // The request URI stays the configured hostname; only the transport target is pinned. + require.Equal(t, testInviteTargetHost, first.req.Recipient.Host) + expectSecond, ok := valid[first.req.Destination()] + require.True(t, ok, "first INVITE went to %s, which pairs a host with another record's port", first.req.Destination()) + + require.NoError(t, first.transaction.SendResponse( + sip.NewResponseFromRequest(first.req, sip.StatusInternalServerError, "Server Internal Error", nil))) + + second := nextTransaction(t, sipClient) + require.Equal(t, sip.INVITE, second.req.Method) + require.Equal(t, expectSecond, second.req.Destination()) + + require.NoError(t, second.transaction.SendResponse( + sip.NewSDPResponseFromRequest(second.req, []byte(testMinimalSDP)))) + + select { + case ack := <-sipClient.requests: + require.Equal(t, sip.ACK, ack.req.Method) + case <-time.After(2 * time.Second): + require.Fail(t, "expected ACK after the second INVITE succeeded") + } +} + +// A status that answers the request must not burn through the other addresses. +func TestOutboundNoFailoverOnTerminalStatus(t *testing.T) { + d := twoSBCDNS(t) + sipClient := startOutboundCall(t, TestClientConfig{DNSResolver: d.resolver()}, MinimalCreateSIPParticipantRequest()) + + first := nextTransaction(t, sipClient) + require.NoError(t, first.transaction.SendResponse( + sip.NewResponseFromRequest(first.req, sip.StatusBusyHere, "Busy Here", nil))) + + select { + case tr := <-sipClient.transactions: + require.Fail(t, "unexpected retry after a terminal status", "method=%v", tr.req.Method) + case <-time.After(500 * time.Millisecond): + } +} + +// With an explicit port SRV is skipped, so every candidate keeps that port. +func TestOutboundExplicitPortSkipsSRV(t *testing.T) { + d := twoSBCDNS(t) + req := MinimalCreateSIPParticipantRequest() + req.Address = testInviteTargetHost + ":5006" + sipClient := startOutboundCall(t, TestClientConfig{DNSResolver: d.resolver()}, req) + + first := nextTransaction(t, sipClient) + require.Contains(t, []string{testSBC1IP + ":5006", testSBC2IP + ":5006"}, first.req.Destination()) + require.Zero(t, d.queries("_sip._udp."+testInviteTargetHost, dnsTypeSRV), + "SRV must not be consulted when the trunk address carries a port") + // The pinned destination is an IP literal, so sipgo does not resolve again. + require.Equal(t, 1, d.queries(testInviteTargetHost, dnsTypeA), + "the host should be resolved exactly once, by ResolveTargets") + + require.NoError(t, first.transaction.SendResponse( + sip.NewResponseFromRequest(first.req, sip.StatusInternalServerError, "Server Internal Error", nil))) + + second := nextTransaction(t, sipClient) + require.NotEqual(t, first.req.Destination(), second.req.Destination()) + require.Contains(t, []string{testSBC1IP + ":5006", testSBC2IP + ":5006"}, second.req.Destination()) +} diff --git a/pkg/sip/outbound_utilities_test.go b/pkg/sip/outbound_utilities_test.go index 785508075..7d4744c06 100644 --- a/pkg/sip/outbound_utilities_test.go +++ b/pkg/sip/outbound_utilities_test.go @@ -18,6 +18,8 @@ import ( "context" "errors" "fmt" + "net" + "net/netip" "sync/atomic" "testing" "time" @@ -356,6 +358,13 @@ type testSIPClient struct { sequence uint64 } +func (w *testSIPClient) ResolveTargets(ctx context.Context, network, host string, port int, sipScheme string) ([]netip.AddrPort, error) { + if w.client == nil { + return nil, errors.New("no sip client") + } + return w.client.ResolveTargets(ctx, network, host, port, sipScheme) +} + func (w *testSIPClient) FillRequestBlanks(req *sip.Request) { sipgo.ClientRequestAddVia(w.client, req) if req.From() == nil { @@ -480,6 +489,7 @@ type TestClientConfig struct { GetSipClient GetSipClientFunc // NewTestClientFunc if nil GetRoom GetRoomFunc // newTestRoom if nil Handler Handler // empty TestHandler if nil + DNSResolver *net.Resolver // system resolver if nil } func NewOutboundTestClient(t testing.TB, cfg TestClientConfig) *Client { @@ -557,7 +567,17 @@ func NewOutboundTestClient(t testing.TB, cfg TestClientConfig) *Client { MediaIP: localIP, } - err = client.Start(nil, sconf) // needed to set sconf + var agent *sipgo.UserAgent + if cfg.DNSResolver != nil { + agent, err = sipgo.NewUA( + sipgo.WithUserAgent(UserAgent), + sipgo.WithUserAgentDNSResolver(cfg.DNSResolver), + ) + if err != nil { + t.Fatalf("failed to create user agent: %v", err) + } + } + err = client.Start(agent, sconf) // needed to set sconf if err != nil { t.Fatalf("failed to start client: %v", err) }