inProgressInvite lifecycle, stable ID generation, minor log field reshuffling

Author: alexlivekit

pkg/sip/inbound.go

@@ -30,6 +30,7 @@ import (
 
 	msdk "github.com/livekit/media-sdk"
 	"github.com/livekit/protocol/rpc"
+	uuid "github.com/satori/go.uuid"
 
 	"github.com/frostbyte73/core"
 	"github.com/icholy/digest"
@@ -64,6 +65,8 @@ 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,20 +140,23 @@ func (s *Server) getCallInfo(id string) *inboundCallInfo {
 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]
-		}
-	}
-	if len(s.inProgressInvites) >= digestLimit {
-		s.inProgressInvites = s.inProgressInvites[1:]
+	is, ok := s.inProgressInvites[sipCallID]
+	if ok {
+		return is
 	}
-	is := &inProgressInvite{sipCallID: sipCallID}
-	s.inProgressInvites = append(s.inProgressInvites, is)
+	is = &inProgressInvite{sipCallID: sipCallID}
+	s.inProgressInvites[sipCallID] = is
+
+	go func() {
+		time.Sleep(inviteCredentialValidity)
+		s.imu.Lock()
+		defer s.imu.Unlock()
+		delete(s.inProgressInvites, sipCallID)
+	}()
 	return is
 }
 
-func (s *Server) handleInviteAuth(log logger.Logger, req *sip.Request, tx sip.ServerTransaction, from, username, password string) (ok bool) {
+func (s *Server) handleInviteAuth(log logger.Logger, req *sip.Request, tx sip.ServerTransaction, from, username, password string, inviteState *inProgressInvite) (ok bool) {
 	log = log.WithValues(
 		"username", username,
 		"passwordHash", hashPassword(password),
@@ -178,8 +184,6 @@ func (s *Server) handleInviteAuth(log logger.Logger, req *sip.Request, tx sip.Se
 	}
 	ci := s.getCallInfo(sipCallID)
 	ci.countInvite(log, req)
-	inviteState := s.getInvite(sipCallID)
-	log = log.WithValues("inviteStateSipCallID", sipCallID)
 
 	h := req.GetHeader("Proxy-Authorization")
 	if h == nil {
@@ -220,7 +224,6 @@ func (s *Server) handleInviteAuth(log logger.Logger, req *sip.Request, tx sip.Se
 	// 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))
@@ -295,18 +298,18 @@ 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"))
 	}
-	callID := lksip.NewCallID()
+	sipCallID := legCallIDFromReq(req)
 	tr := callTransportFromReq(req)
 	legTr := legTransportFromReq(req)
 	log := s.log.WithValues(
-		"callID", callID,
+		"sipCallID", sipCallID,
 		"fromIP", src.Addr(),
 		"toIP", req.Destination(),
 		"transport", tr,
 	)
 
 	var call *inboundCall
-	cc := s.newInbound(log, LocalTag(callID), s.ContactURI(legTr), req, tx, func(headers map[string]string) map[string]string {
+	cc := s.newInbound(log, "unassigned", s.ContactURI(legTr), req, tx, func(headers map[string]string) map[string]string {
 		c := call
 		if c == nil || len(c.attrsToHdr) == 0 {
 			return headers
@@ -319,8 +322,6 @@ 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 {
 		if s.conf.HideInboundPort {
@@ -330,6 +331,28 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		}
 		return psrpc.NewError(psrpc.InvalidArgument, errors.Wrap(err, "invite validation failed"))
 	}
+
+	// Establish ID
+	if _, ok := req.To().Params.Get("tag"); !ok {
+		// No to-tag on the invite means we need to generate one per RFC 3261 section 12.
+		if !inviteHasAuth(req) {
+			// No auth = a 407 response and another INVITE+auth.
+			// Generate a new to-tag early, to make sure both INVITES have the same ID.
+			uuid, _ := uuid.NewV4() // Same as NewResponseFromRequest in sipgo
+			req.To().Params.Add("tag", uuid.String())
+		}
+	}
+	inviteProgress := s.getInvite(req.CallID().Value())
+	callID := inviteProgress.lkCallID
+	if callID == "" {
+		callID = lksip.NewCallID()
+		inviteProgress.lkCallID = callID
+	}
+
+	log = log.WithValues("callID", callID)
+	cc.log = log
+	log.Infow("processing invite")
+
 	ctx, span := tracer.Start(ctx, "Server.onInvite")
 	defer span.End()
 
@@ -352,12 +375,6 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		cc.Processing()
 	}
 
-	// Extract SIP Call ID directly from the request
-	sipCallID := ""
-	if h := req.CallID(); h != nil {
-		sipCallID = h.Value()
-	}
-
 	callInfo := &rpc.SIPCall{
 		LkCallId:  callID,
 		SipCallId: sipCallID,
@@ -421,7 +438,7 @@ 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()
 		}
-		if !s.handleInviteAuth(log, req, tx, from.User, r.Username, r.Password) {
+		if !s.handleInviteAuth(log, req, tx, from.User, r.Username, r.Password, inviteProgress) {
 			cmon.InviteErrorShort("unauthorized")
 			// handleInviteAuth will generate the SIP Response as needed
 			return psrpc.NewErrorf(psrpc.PermissionDenied, "invalid credentials were provided")

pkg/sip/protocol.go

@@ -173,6 +173,13 @@ 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

@@ -43,8 +43,7 @@ import (
 )
 
 const (
-	UserAgent   = "LiveKit"
-	digestLimit = 500
+	UserAgent = "LiveKit"
 )
 
 const (
@@ -133,7 +132,7 @@ type Server struct {
 	sipUnhandled RequestHandler
 
 	imu               sync.Mutex
-	inProgressInvites []*inProgressInvite
+	inProgressInvites map[string]*inProgressInvite
 
 	closing     core.Fuse
 	cmu         sync.RWMutex
@@ -155,20 +154,22 @@ type Server struct {
 type inProgressInvite struct {
 	sipCallID string
 	challenge digest.Challenge
+	lkCallID  string // SCL_* LiveKit call ID assigned to this dialog
 }
 
 func NewServer(region string, conf *config.Config, log logger.Logger, mon *stats.Monitor, getIOClient GetIOInfoClient) *Server {
 	if log == nil {
 		log = logger.GetLogger()
 	}
 	s := &Server{
-		log:         log,
-		conf:        conf,
-		region:      region,
-		mon:         mon,
-		getIOClient: getIOClient,
-		activeCalls: make(map[RemoteTag]*inboundCall),
-		byLocal:     make(map[LocalTag]*inboundCall),
+		log:               log,
+		conf:              conf,
+		region:            region,
+		mon:               mon,
+		getIOClient:       getIOClient,
+		inProgressInvites: make(map[string]*inProgressInvite),
+		activeCalls:       make(map[RemoteTag]*inboundCall),
+		byLocal:           make(map[LocalTag]*inboundCall),
 	}
 	s.infos.byCallID = expirable.NewLRU[string, *inboundCallInfo](maxCallCache, nil, callCacheTTL)
 	s.initMediaRes()