ai/live: use webhookUrl query param to override auth webhook url

cmd/livepeer/starter/flags.go

@@ -66,7 +66,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	// Live AI:
 	cfg.MediaMTXApiPassword = fs.String("mediaMTXApiPassword", "", "HTTP basic auth password for MediaMTX API requests")
 	cfg.LiveAITrickleHostForRunner = fs.String("liveAITrickleHostForRunner", "", "Trickle Host used by AI Runner; It's used to overwrite the publicly available Trickle Host")
-	cfg.LiveAIAuthApiKey = fs.String("liveAIAuthApiKey", "", "API key to use for Live AI authentication requests")
+	cfg.LiveAIAuthSecret = fs.String("liveAIAuthSecret", "", "HMAC secret for Live AI auth signing")
 	cfg.LiveAIHeartbeatURL = fs.String("liveAIHeartbeatURL", "", "Base URL for Live AI heartbeat requests")
 	cfg.LiveAIHeartbeatHeaders = fs.String("liveAIHeartbeatHeaders", "", "Map of headers to use for Live AI heartbeat requests. e.g. 'header:val,header2:val2'")
 	cfg.LiveAIHeartbeatInterval = fs.Duration("liveAIHeartbeatInterval", *cfg.LiveAIHeartbeatInterval, "Interval to send Live AI heartbeat requests")
@@ -173,4 +173,4 @@ func UpdateNilsForUnsetFlags(cfg LivepeerConfig) LivepeerConfig {
 	}
 
 	return res
-}
+}

cmd/livepeer/starter/starter.go

@@ -176,7 +176,7 @@ type LivepeerConfig struct {
 	KafkaPassword              *string
 	KafkaGatewayTopic          *string
 	MediaMTXApiPassword        *string
-	LiveAIAuthApiKey           *string
+	LiveAIAuthSecret           *string
 	LiveAIHeartbeatURL         *string
 	LiveAIHeartbeatHeaders     *string
 	LiveAIHeartbeatInterval    *time.Duration
@@ -1676,8 +1676,8 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	if cfg.MediaMTXApiPassword != nil {
 		n.MediaMTXApiPassword = *cfg.MediaMTXApiPassword
 	}
-	if cfg.LiveAIAuthApiKey != nil {
-		n.LiveAIAuthApiKey = *cfg.LiveAIAuthApiKey
+	if cfg.LiveAIAuthSecret != nil {
+		n.LiveAIAuthSecret = *cfg.LiveAIAuthSecret
 	}
 	if cfg.LiveAIHeartbeatURL != nil {
 		n.LiveAIHeartbeatURL = *cfg.LiveAIHeartbeatURL

core/livepeernode.go

@@ -159,7 +159,7 @@ type LivepeerNode struct {
 
 	MediaMTXApiPassword        string
 	LiveAITrickleHostForRunner string
-	LiveAIAuthApiKey           string
+	LiveAIAuthSecret           string
 	LiveAIHeartbeatURL         string
 	LiveAIHeartbeatHeaders     map[string]string
 	LiveAIHeartbeatInterval    time.Duration

server/ai_mediaserver.go

@@ -505,8 +505,15 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 		mediaMTXClient := media.NewMediaMTXClient(remoteHost, ls.mediaMTXApiPassword, sourceID, sourceType)
 
 		whepURL := generateWhepUrl(streamName, requestID)
-		if LiveAIAuthWebhookURL != nil {
-			authResp, err := authenticateAIStream(LiveAIAuthWebhookURL, ls.liveAIAuthApiKey, AIAuthRequest{
+		authURL := LiveAIAuthWebhookURL
+		override := qp.Get("webhookUrl")
+		if override != "" {
+			if parsed, err := url.Parse(override); err == nil && parsed.Scheme != "" && parsed.Host != "" {
+				authURL = parsed
+			}
+		}
+		if authURL != nil {
+			authResp, err := authenticateAIStream(authURL, ls.liveAIAuthSecret, AIAuthRequest{
 				Stream:      streamName,
 				Type:        sourceTypeStr,
 				QueryParams: queryParams,
@@ -977,8 +984,15 @@ func (ls *LivepeerServer) CreateWhip(server *media.WHIPServer) http.Handler {
 
 			ctx = clog.AddVal(ctx, "source_type", sourceTypeStr)
 
-			if LiveAIAuthWebhookURL != nil {
-				authResp, err := authenticateAIStream(LiveAIAuthWebhookURL, ls.liveAIAuthApiKey, AIAuthRequest{
+			authURL := LiveAIAuthWebhookURL
+			override := r.URL.Query().Get("webhookUrl")
+			if override != "" {
+				if parsed, err := url.Parse(override); err == nil && parsed.Scheme != "" && parsed.Host != "" {
+					authURL = parsed
+				}
+			}
+			if authURL != nil {
+				authResp, err := authenticateAIStream(authURL, ls.liveAIAuthSecret, AIAuthRequest{
 					Stream:      streamName,
 					Type:        sourceTypeStr,
 					QueryParams: queryParams,

server/auth.go

@@ -10,6 +10,10 @@
 	"net/url"
 	"time"
 
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+
 	"github.com/golang/glog"
 	"github.com/livepeer/go-livepeer/monitor"
 )
@@ -133,7 +137,7 @@
 	paramsMap      map[string]interface{} // unmarshaled params
 }
 
-func authenticateAIStream(authURL *url.URL, apiKey string, req AIAuthRequest) (*AIAuthResponse, error) {
+func authenticateAIStream(authURL *url.URL, secret string, req AIAuthRequest) (*AIAuthResponse, error) {
 	req.StreamKey = req.Stream
 	if authURL == nil {
 		return nil, fmt.Errorf("No auth URL configured")
@@ -151,10 +155,16 @@
 	}
 
 	request.Header.Set("Content-Type", "application/json")
-	request.Header.Set("x-api-key", apiKey)
-	request.Header.Set("Authorization", apiKey)
+
+	ts := time.Now().UTC().Format(time.RFC3339Nano)
+	m := hmac.New(sha256.New, []byte(secret))
+	m.Write([]byte(ts))
+	m.Write(jsonValue)
+	sig := hex.EncodeToString(m.Sum(nil))
+	request.Header.Set("x-timestamp", ts)
+	request.Header.Set("x-signature", sig)
 
 	resp, err := http.DefaultClient.Do(request)
 	if err != nil {
 		return nil, err
 	}

server/mediaserver.go

@@ -127,7 +127,7 @@ type LivepeerServer struct {
 	serverLock        *sync.RWMutex
 
 	mediaMTXApiPassword string
-	liveAIAuthApiKey    string
+	liveAIAuthSecret    string
 	livePaymentInterval time.Duration
 	outSegmentTimeout   time.Duration
 }
@@ -196,7 +196,7 @@ func NewLivepeerServer(ctx context.Context, rtmpAddr string, lpNode *core.Livepe
 		recordingsAuthResponses: cache.New(time.Hour, 2*time.Hour),
 		AISessionManager:        NewAISessionManager(lpNode, AISessionManagerTTL),
 		mediaMTXApiPassword:     lpNode.MediaMTXApiPassword,
-		liveAIAuthApiKey:        lpNode.LiveAIAuthApiKey,
+		liveAIAuthSecret:        lpNode.LiveAIAuthSecret,
 		livePaymentInterval:     lpNode.LivePaymentInterval,
 		outSegmentTimeout:       lpNode.LiveOutSegmentTimeout,
 	}