feat(remote_signer): Add job token signing endpoint for byoc job types

CHANGELOG_PENDING.md

@@ -4,6 +4,8 @@
 
 ### Breaking Changes 🚨🚨
 
+- BYOC orchestrator job requests now hard-reject invalid non-empty `Livepeer-Payment` headers with `HTTP 400 Bad Request` (`Could not parse payment`) instead of logging and continuing until balance depletion. Gateways/SDKs must omit stale payment headers or send a valid fresh payment.
+
 ### Features ⚒
 
 #### General

byoc/job_orchestrator.go

@@ -491,6 +491,10 @@ func (bso *BYOCOrchestratorServer) confirmPayment(ctx context.Context, sender et
 		orchBal, pmtErr := bso.processPayment(ctx, sender, capability, paymentHdr)
 		if pmtErr != nil {
 			//log if there are payment errors but continue, balance will runout and clean up
+			if paymentHdr != "" {
+				clog.Errorf(ctx, "rejecting request: payment header present but invalid: %v", pmtErr)
+				return errPaymentError
+			}
 			clog.Infof(ctx, "job payment error: %v", pmtErr)
 		}
 
@@ -502,25 +506,22 @@ func (bso *BYOCOrchestratorServer) confirmPayment(ctx context.Context, sender et
 	return nil
 }
 
-// process payment and return balance
+// processPayment decodes and applies the payment header if present.
+// Always returns a non-nil balance so callers can safely compare.
 func (bso *BYOCOrchestratorServer) processPayment(ctx context.Context, sender ethcommon.Address, capability string, paymentHdr string) (*big.Rat, error) {
 	if paymentHdr != "" {
 		payment, err := getPayment(paymentHdr)
 		if err != nil {
 			clog.Errorf(ctx, "job payment invalid: %v", err)
-			return nil, errPaymentError
+			return bso.getPaymentBalance(sender, capability), errPaymentError
 		}
 
 		if err := bso.orch.ProcessPayment(ctx, payment, core.ManifestID(capability)); err != nil {
-			bso.orch.FreeExternalCapabilityCapacity(capability)
 			clog.Errorf(ctx, "Error processing payment: %v", err)
-			return nil, errPaymentError
+			return bso.getPaymentBalance(sender, capability), errPaymentError
 		}
 	}
-	orchBal := bso.getPaymentBalance(sender, capability)
-
-	return orchBal, nil
-
+	return bso.getPaymentBalance(sender, capability), nil
 }
 
 func (bso *BYOCOrchestratorServer) chargeForCompute(start time.Time, price *net.PriceInfo, sender ethcommon.Address, jobId string) {
@@ -566,18 +567,26 @@ func (bso *BYOCOrchestratorServer) verifyJobCreds(ctx context.Context, jobCreds
 		return nil, errSegSig
 	}
 
-	if !bso.orch.VerifySig(ethcommon.HexToAddress(jobData.Sender), jobData.Request+jobData.Parameters, sigByte) {
-		clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender)
-		return nil, errSegSig
-	}
+	sender := ethcommon.HexToAddress(jobData.Sender)
 
-	if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil {
-		return nil, errZeroCapacity
+	// Verify V1 structured binary format
+	v1Payload := FlattenBYOCJob(&BYOCJobSigningInput{
+		ID:             jobData.ID,
+		Capability:     jobData.Capability,
+		Request:        jobData.Request,
+		Parameters:     jobData.Parameters,
+		TimeoutSeconds: jobData.Timeout,
+	})
+	if bso.orch.VerifySig(sender, string(v1Payload), sigByte) {
+		if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil {
+			return nil, errZeroCapacity
+		}
+		jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability)
+		return jobData, nil
 	}
 
-	jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability)
-
-	return jobData, nil
+	clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender)
+	return nil, errSegSig
 }
 
 func (bso *BYOCOrchestratorServer) verifyTokenCreds(ctx context.Context, tokenCreds string) (*JobSender, error) {

byoc/types.go

@@ -3,7 +3,10 @@ package byoc
 import (
 	"context"
 	"crypto/tls"
+	"encoding/binary"
 	"errors"
+	"fmt"
+	"math"
 	"math/big"
 	gonet "net"
 	"net/http"
@@ -279,3 +282,81 @@ type byocLiveRequestParams struct {
 	// when the write for the last segment started
 	lastSegmentTime time.Time
 }
+
+// BYOCJobSigV1Prefix is the 16-byte domain separator for BYOC job signatures (V1).
+// Prevents cross-protocol signature replay. Changing this prefix is wire-breaking.
+const BYOCJobSigV1Prefix = "LP_BYOC_JOB_V1\x00\x00"
+
+// MaxBYOCJobTimeoutSeconds is the largest timeout_seconds value accepted on the wire.
+// The V1 format encodes timeout as a big-endian uint32.
+const MaxBYOCJobTimeoutSeconds = math.MaxUint32
+
+// ValidateBYOCJobTimeoutSeconds rejects values that cannot be encoded in the V1 wire format.
+func ValidateBYOCJobTimeoutSeconds(seconds int) error {
+	if seconds <= 0 {
+		return errors.New("timeout_seconds must be positive")
+	}
+	if seconds > MaxBYOCJobTimeoutSeconds {
+		return fmt.Errorf("timeout_seconds exceeds maximum %d", MaxBYOCJobTimeoutSeconds)
+	}
+	return nil
+}
+
+// BYOCJobSigningInput holds the fields that are bound into a BYOC job signature.
+type BYOCJobSigningInput struct {
+	ID             string
+	Capability     string
+	Request        string
+	Parameters     string
+	TimeoutSeconds int
+}
+
+// FlattenBYOCJob produces a deterministic binary representation of a BYOC job
+// for signing, similar to SegTranscodingMetadata.Flatten() used by LV2V.
+//
+// Wire format:
+//
+//	version(16) || timeout(4,BE uint32) || len(id)(4,BE) || id || len(cap)(4,BE) || cap
+//	           || len(req)(4,BE) || req || len(params)(4,BE) || params
+func FlattenBYOCJob(job *BYOCJobSigningInput) []byte {
+	idBytes := []byte(job.ID)
+	capBytes := []byte(job.Capability)
+	reqBytes := []byte(job.Request)
+	paramsBytes := []byte(job.Parameters)
+
+	size := 16 + 4 +
+		4 + len(idBytes) +
+		4 + len(capBytes) +
+		4 + len(reqBytes) +
+		4 + len(paramsBytes)
+
+	buf := make([]byte, size)
+	offset := 0
+
+	copy(buf[offset:], []byte(BYOCJobSigV1Prefix))
+	offset += 16
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(job.TimeoutSeconds))
+	offset += 4
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(idBytes)))
+	offset += 4
+	copy(buf[offset:], idBytes)
+	offset += len(idBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(capBytes)))
+	offset += 4
+	copy(buf[offset:], capBytes)
+	offset += len(capBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(reqBytes)))
+	offset += 4
+	copy(buf[offset:], reqBytes)
+	offset += len(reqBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(paramsBytes)))
+	offset += 4
+	copy(buf[offset:], paramsBytes)
+
+	return buf
+}

byoc/types_test.go

@@ -0,0 +1,26 @@
+package byoc
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateBYOCJobTimeoutSeconds(t *testing.T) {
+	require := require.New(t)
+
+	require.NoError(ValidateBYOCJobTimeoutSeconds(1))
+	require.NoError(ValidateBYOCJobTimeoutSeconds(MaxBYOCJobTimeoutSeconds))
+
+	err := ValidateBYOCJobTimeoutSeconds(0)
+	require.Error(err)
+	require.Contains(err.Error(), "positive")
+
+	err = ValidateBYOCJobTimeoutSeconds(-1)
+	require.Error(err)
+	require.Contains(err.Error(), "positive")
+
+	err = ValidateBYOCJobTimeoutSeconds(MaxBYOCJobTimeoutSeconds + 1)
+	require.Error(err)
+	require.Contains(err.Error(), "maximum")
+}

byoc/utils.go

@@ -33,7 +33,20 @@ var sendJobReqWithTimeout = sendReqWithTimeout
 func (g *gatewayJob) sign() error {
 	//sign the request
 	gateway := g.node.OrchestratorPool.Broadcaster()
-	sig, err := gateway.Sign([]byte(g.Job.Req.Request + g.Job.Req.Parameters))
+
+	if err := ValidateBYOCJobTimeoutSeconds(g.Job.Req.Timeout); err != nil {
+		return err
+	}
+
+	sigPayload := FlattenBYOCJob(&BYOCJobSigningInput{
+		ID:             g.Job.Req.ID,
+		Capability:     g.Job.Req.Capability,
+		Request:        g.Job.Req.Request,
+		Parameters:     g.Job.Req.Parameters,
+		TimeoutSeconds: g.Job.Req.Timeout,
+	})
+
+	sig, err := gateway.Sign(sigPayload)
 	if err != nil {
 		return errors.New(fmt.Sprintf("Unable to sign request err=%v", err))
 	}
@@ -67,6 +80,9 @@ func parseJobRequest(jobReq string) (*JobRequest, error) {
 	if jobData.Timeout == 0 {
 		return nil, errNoTimeoutSet
 	}
+	if err := ValidateBYOCJobTimeoutSeconds(jobData.Timeout); err != nil {
+		return nil, err
+	}
 
 	return &jobData, nil
 }

core/orchestrator.go

@@ -418,9 +418,18 @@ func (orch *orchestrator) priceInfo(sender ethcommon.Address, manifestID Manifes
 					continue
 				}
 				for modelID := range constraints.Models {
-					price := orch.node.GetBasePriceForCap(sender.String(), Capability(cap), modelID)
-					if price == nil {
-						price = orch.node.GetBasePriceForCap("default", Capability(cap), modelID)
+					var price *big.Rat
+					if Capability(cap) == Capability_BYOC {
+						// BYOC prices are stored in jobPriceInfo, keyed by capability name
+						price = orch.node.GetPriceForJob(sender.String(), modelID)
+						if price == nil || price.Sign() == 0 {
+							price = orch.node.GetPriceForJob("default", modelID)
+						}
+					} else {
+						price = orch.node.GetBasePriceForCap(sender.String(), Capability(cap), modelID)
+						if price == nil {
+							price = orch.node.GetBasePriceForCap("default", Capability(cap), modelID)
+						}
 					}
 
 					if price != nil {