Revert "Reuse the same ID for both auth-less and auth-ful INVITEs" (#533)

Author: dennwc

pkg/sip/inbound.go

@@ -42,7 +42,6 @@ import (
 	"github.com/livekit/protocol/rpc"
 	lksip "github.com/livekit/protocol/sip"
 	"github.com/livekit/protocol/tracer"
-	"github.com/livekit/protocol/utils"
 	"github.com/livekit/protocol/utils/traceid"
 	"github.com/livekit/psrpc"
 	lksdk "github.com/livekit/server-sdk-go/v2"
@@ -65,8 +64,6 @@ const (
 	inviteOKRetryAttempts      = 5
 	inviteOKRetryAttemptsNoACK = 2
 	inviteOkAckLateTimeout     = inviteOkRetryIntervalMax
-
-	inviteCredentialValidity = 60 * time.Minute // Allow reuse of credentials for 1h
 )
 
 var errNoACK = errors.New("no ACK received for 200 OK")
@@ -137,50 +134,23 @@ func (s *Server) getCallInfo(id string) *inboundCallInfo {
 	return c
 }
 
-func (s *Server) cleanupInvites() {
-	ticker := time.NewTicker(5 * time.Minute) // Periodic cleanup every 5 minutes
-	defer ticker.Stop()
-	for {
-		select {
-		case <-s.closing.Watch():
-			return
-		case <-ticker.C:
-			s.imu.Lock()
-			for it := s.inviteTimeoutQueue.IterateRemoveAfter(inviteCredentialValidity); it.Next(); {
-				key := it.Item().Value
-				delete(s.inProgressInvites, key)
-			}
-			s.imu.Unlock()
+func (s *Server) getInvite(sipCallID string) *inProgressInvite {
+	s.imu.Lock()
+	defer s.imu.Unlock()
+	for i := range s.inProgressInvites {
+		if s.inProgressInvites[i].sipCallID == sipCallID {
+			return s.inProgressInvites[i]
 		}
 	}
-}
-
-func (s *Server) getInvite(sipCallID, toTag, fromTag string) *inProgressInvite {
-	key := dialogKey{
-		sipCallID: sipCallID,
-		toTag:     toTag,
-		fromTag:   fromTag,
-	}
-
-	s.imu.RLock()
-	is, exists := s.inProgressInvites[key]
-	s.imu.RUnlock()
-	if !exists {
-		s.imu.Lock()
-		is, exists = s.inProgressInvites[key]
-		if !exists {
-			is = &inProgressInvite{sipCallID: sipCallID, timeoutLink: utils.TimeoutQueueItem[dialogKey]{Value: key}}
-			s.inProgressInvites[key] = is
-		}
-		s.imu.Unlock()
+	if len(s.inProgressInvites) >= digestLimit {
+		s.inProgressInvites = s.inProgressInvites[1:]
 	}
-
-	// Always reset the timeout link, whether just created or not
-	s.inviteTimeoutQueue.Reset(&is.timeoutLink)
+	is := &inProgressInvite{sipCallID: sipCallID}
+	s.inProgressInvites = append(s.inProgressInvites, is)
 	return is
 }
 
-func (s *Server) handleInviteAuth(tid traceid.ID, log logger.Logger, req *sip.Request, tx sip.ServerTransaction, from, username, password string, inviteState *inProgressInvite) (ok bool) {
+func (s *Server) handleInviteAuth(tid traceid.ID, log logger.Logger, req *sip.Request, tx sip.ServerTransaction, from, username, password string) (ok bool) {
 	log = log.WithValues(
 		"username", username,
 		"passwordHash", hashPassword(password),
@@ -201,6 +171,14 @@ func (s *Server) handleInviteAuth(tid traceid.ID, log logger.Logger, req *sip.Re
 		_ = tx.Respond(sip.NewResponseFromRequest(req, 100, "Processing", nil))
 	}
 
+	// Extract SIP Call ID for tracking in-progress invites
+	sipCallID := ""
+	if h := req.CallID(); h != nil {
+		sipCallID = h.Value()
+	}
+	inviteState := s.getInvite(sipCallID)
+	log = log.WithValues("inviteStateSipCallID", sipCallID)
+
 	h := req.GetHeader("Proxy-Authorization")
 	if h == nil {
 		inviteState.challenge = digest.Challenge{
@@ -252,6 +230,7 @@ func (s *Server) handleInviteAuth(tid traceid.ID, log logger.Logger, req *sip.Re
 	// Check if we have a valid challenge state
 	if inviteState.challenge.Realm == "" {
 		log.Warnw("No challenge state found for authentication attempt", errors.New("missing challenge state"),
+			"sipCallID", sipCallID,
 			"expectedRealm", UserAgent,
 		)
 		_ = tx.Respond(sip.NewResponseFromRequest(req, 401, "Bad credentials", nil))
@@ -326,18 +305,20 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		s.log.Errorw("cannot parse source IP", err, "fromIP", src)
 		return psrpc.NewError(psrpc.MalformedRequest, errors.Wrap(err, "cannot parse source IP"))
 	}
-	sipCallID := legCallIDFromReq(req)
+	callID := lksip.NewCallID()
+	tid := traceid.FromGUID(callID)
 	tr := callTransportFromReq(req)
 	legTr := legTransportFromReq(req)
 	log := s.log.WithValues(
-		"sipCallID", sipCallID,
+		"callID", callID,
+		"traceID", tid.String(),
 		"fromIP", src.Addr(),
 		"toIP", req.Destination(),
 		"transport", tr,
 	)
 
 	var call *inboundCall
-	cc := s.newInbound(log, s.ContactURI(legTr), req, tx, func(headers map[string]string) map[string]string {
+	cc := s.newInbound(log, LocalTag(callID), s.ContactURI(legTr), req, tx, func(headers map[string]string) map[string]string {
 		c := call
 		if c == nil || len(c.attrsToHdr) == 0 {
 			return headers
@@ -350,53 +331,25 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 	})
 	log = LoggerWithParams(log, cc)
 	log = LoggerWithHeaders(log, cc)
+	cc.log = log
+	log.Infow("processing invite")
 
 	if err := cc.ValidateInvite(); err != nil {
-		log.Errorw("invalid invite", err)
 		if s.conf.HideInboundPort {
 			cc.Drop()
 		} else {
 			cc.RespondAndDrop(sip.StatusBadRequest, "Bad request")
 		}
 		return psrpc.NewError(psrpc.InvalidArgument, errors.Wrap(err, "invite validation failed"))
 	}
-
-	// Establish ID
-	fromTag, _ := req.From().Params.Get("tag") // always exists, via ValidateInvite() check
-	toParams := req.To().Params                // To() always exists, via ValidateInvite() check
-	if toParams == nil {
-		toParams = sip.NewParams()
-		req.To().Params = toParams
-	}
-	toTag, ok := toParams.Get("tag")
-	if !ok {
-		// No to-tag on the invite means we need to generate one per RFC 3261 section 12.
-		// Generate a new to-tag early, to make sure both INVITES have the same ID.
-		toTag = utils.NewGuid("")
-		toParams.Add("tag", toTag)
-	}
-	inviteProgress := s.getInvite(sipCallID, toTag, fromTag)
-	callID := inviteProgress.lkCallID
-	if callID == "" {
-		callID = lksip.NewCallID()
-		inviteProgress.lkCallID = callID
-	}
-	cc.id = LocalTag(callID)
-	tid := traceid.FromGUID(sipCallID)
-
-	log = log.WithValues("callID", callID)
-	log = log.WithValues("traceID", tid.String())
-	cc.log = log
-	log.Infow("processing invite")
-
 	ctx, span := tracer.Start(ctx, "Server.onInvite")
 	defer span.End()
 
 	s.cmu.RLock()
-	existing := s.byCallID[sipCallID]
+	existing := s.byCallID[cc.SIPCallID()]
 	s.cmu.RUnlock()
 	if existing != nil && existing.cc.InviteCSeq() < cc.InviteCSeq() {
-		log.Infow("accepting reinvite", "content-type", req.ContentType(), "content-length", req.ContentLength())
+		log.Infow("accepting reinvite", "sipCallID", existing.cc.ID(), "content-type", req.ContentType(), "content-length", req.ContentLength())
 		existing.log().Infow("reinvite", "content-type", req.ContentType(), "content-length", req.ContentLength(), "cseq", cc.InviteCSeq())
 		cc.AcceptAsKeepAlive(existing.cc.OwnSDP())
 		return nil
@@ -423,7 +376,7 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 
 	callInfo := &rpc.SIPCall{
 		LkCallId:  callID,
-		SipCallId: sipCallID,
+		SipCallId: cc.SIPCallID(),
 		SourceIp:  src.Addr().String(),
 		Address:   ToSIPUri("", cc.Address()),
 		From:      ToSIPUri("", from),
@@ -494,15 +447,15 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 			// We will send password request anyway, so might as well signal that the progress is made.
 			cc.Processing()
 		}
-		s.getCallInfo(sipCallID).countInvite(log, req)
-		if !s.handleInviteAuth(tid, log, req, tx, from.User, r.Username, r.Password, inviteProgress) {
+		s.getCallInfo(cc.SIPCallID()).countInvite(log, req)
+		if !s.handleInviteAuth(tid, log, req, tx, from.User, r.Username, r.Password) {
 			cmon.InviteErrorShort("unauthorized")
 			// handleInviteAuth will generate the SIP Response as needed
 			return psrpc.NewErrorf(psrpc.PermissionDenied, "invalid credentials were provided")
 		}
 		// ok
 	case AuthAccept:
-		s.getCallInfo(sipCallID).countInvite(log, req)
+		s.getCallInfo(cc.SIPCallID()).countInvite(log, req)
 		// ok
 	}
 
@@ -1422,10 +1375,11 @@ func (c *inboundCall) transferCall(ctx context.Context, transferTo string, heade
 
 }
 
-func (s *Server) newInbound(log logger.Logger, contact URI, invite *sip.Request, inviteTx sip.ServerTransaction, getHeaders setHeadersFunc) *sipInbound {
+func (s *Server) newInbound(log logger.Logger, id LocalTag, contact URI, invite *sip.Request, inviteTx sip.ServerTransaction, getHeaders setHeadersFunc) *sipInbound {
 	c := &sipInbound{
 		log:      log,
 		s:        s,
+		id:       id,
 		invite:   invite,
 		inviteTx: inviteTx,
 		legTr:    legTransportFromReq(invite),

pkg/sip/outbound.go

@@ -852,7 +852,7 @@ authLoop:
 		if err != nil {
 			return nil, fmt.Errorf("invalid challenge %q: %w", challengeStr, err)
 		}
-		toHeader = resp.To()
+		toHeader := resp.To()
 		if toHeader == nil {
 			return nil, errors.New("no 'To' header on Response")
 		}

pkg/sip/protocol.go

@@ -173,13 +173,6 @@ func legTransportFromReq(req *sip.Request) Transport {
 	return ""
 }
 
-func legCallIDFromReq(req *sip.Request) string {
-	if callID := req.CallID(); callID != nil {
-		return callID.Value()
-	}
-	return ""
-}
-
 func transportPort(c *config.Config, t Transport) int {
 	if t == TransportTLS {
 		if tc := c.TLS; tc != nil {

pkg/sip/server.go

@@ -35,7 +35,6 @@ import (
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/rpc"
-	"github.com/livekit/protocol/utils"
 	"github.com/livekit/protocol/utils/traceid"
 	"github.com/livekit/sipgo"
 	"github.com/livekit/sipgo/sip"
@@ -45,7 +44,8 @@ import (
 )
 
 const (
-	UserAgent = "LiveKit"
+	UserAgent   = "LiveKit"
+	digestLimit = 500
 )
 
 const (
@@ -127,25 +127,18 @@ type Handler interface {
 	OnSessionEnd(ctx context.Context, callIdentifier *CallIdentifier, callInfo *livekit.SIPCallInfo, reason string)
 }
 
-type dialogKey struct {
-	sipCallID string
-	toTag     string
-	fromTag   string
-}
-
 type Server struct {
-	log                logger.Logger
-	mon                *stats.Monitor
-	region             string
-	sipSrv             *sipgo.Server
-	getIOClient        GetIOInfoClient
-	getRoom            GetRoomFunc
-	sipListeners       []io.Closer
-	sipUnhandled       RequestHandler
-	inviteTimeoutQueue utils.TimeoutQueue[dialogKey]
-
-	imu               sync.RWMutex
-	inProgressInvites map[dialogKey]*inProgressInvite
+	log          logger.Logger
+	mon          *stats.Monitor
+	region       string
+	sipSrv       *sipgo.Server
+	getIOClient  GetIOInfoClient
+	getRoom      GetRoomFunc
+	sipListeners []io.Closer
+	sipUnhandled RequestHandler
+
+	imu               sync.Mutex
+	inProgressInvites []*inProgressInvite
 
 	closing     core.Fuse
 	cmu         sync.RWMutex
@@ -166,10 +159,8 @@ type Server struct {
 }
 
 type inProgressInvite struct {
-	sipCallID   string
-	challenge   digest.Challenge
-	lkCallID    string // SCL_* LiveKit call ID assigned to this dialog
-	timeoutLink utils.TimeoutQueueItem[dialogKey]
+	sipCallID string
+	challenge digest.Challenge
 }
 
 type ServerOption func(s *Server)
@@ -187,16 +178,15 @@ func NewServer(region string, conf *config.Config, log logger.Logger, mon *stats
 		log = logger.GetLogger()
 	}
 	s := &Server{
-		log:               log,
-		conf:              conf,
-		region:            region,
-		mon:               mon,
-		getIOClient:       getIOClient,
-		getRoom:           DefaultGetRoomFunc,
-		inProgressInvites: make(map[dialogKey]*inProgressInvite),
-		byRemoteTag:       make(map[RemoteTag]*inboundCall),
-		byLocalTag:        make(map[LocalTag]*inboundCall),
-		byCallID:          make(map[string]*inboundCall),
+		log:         log,
+		conf:        conf,
+		region:      region,
+		mon:         mon,
+		getIOClient: getIOClient,
+		getRoom:     DefaultGetRoomFunc,
+		byRemoteTag: make(map[RemoteTag]*inboundCall),
+		byLocalTag:  make(map[LocalTag]*inboundCall),
+		byCallID:    make(map[string]*inboundCall),
 	}
 	for _, option := range options {
 		option(s)
@@ -340,9 +330,6 @@ func (s *Server) Start(agent *sipgo.UserAgent, sc *ServiceConfig, tlsConf *tls.C
 		}
 	}
 
-	// Start the cleanup task
-	go s.cleanupInvites()
-
 	return nil
 }