Check write permission for output locations (#1247)

* Check write permission for output locations

* fix test

* refactor to avoid re-checking locations

* memory location will only parse in go-tools for unit tests

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/lib/pq v1.10.9
 	github.com/livepeer/go-api-client v0.4.23-0.20240426140555-b490b47e4df3
-	github.com/livepeer/go-tools v0.3.6
+	github.com/livepeer/go-tools v0.3.7
 	github.com/livepeer/joy4 v0.1.1
 	github.com/livepeer/livepeer-data v0.8.1
 	github.com/livepeer/m3u8 v0.11.1

go.sum

@@ -447,12 +447,10 @@ github.com/libp2p/go-netroute v0.2.0 h1:0FpsbsvuSnAhXFnCY0VLFbJOzaK0VnP0r1QT/o4n
 github.com/libp2p/go-netroute v0.2.0/go.mod h1:Vio7LTzZ+6hoT4CMZi5/6CpY3Snzh2vgZhWgxMNwlQI=
 github.com/libp2p/go-openssl v0.1.0 h1:LBkKEcUv6vtZIQLVTegAil8jbNpJErQ9AnT+bWV+Ooo=
 github.com/libp2p/go-openssl v0.1.0/go.mod h1:OiOxwPpL3n4xlenjx2h7AwSGaFSC/KZvf6gNdOBQMtc=
-github.com/livepeer/go-api-client v0.4.22 h1:AIb+JkLDHTjj4OaiNdnfkM4DnHnmNlFCyvb2AnH5VJk=
-github.com/livepeer/go-api-client v0.4.22/go.mod h1:Jdb+RI7JyzEZOHd1GUuKofwFDKMO/btTa80SdpUpYQw=
 github.com/livepeer/go-api-client v0.4.23-0.20240426140555-b490b47e4df3 h1:iaPz1ZK2vlH4+zVC6eLES04w+/ly5e8KxX9mBpVzqKQ=
 github.com/livepeer/go-api-client v0.4.23-0.20240426140555-b490b47e4df3/go.mod h1:Jdb+RI7JyzEZOHd1GUuKofwFDKMO/btTa80SdpUpYQw=
-github.com/livepeer/go-tools v0.3.6 h1:LhRnoVVGFCtfBh6WyKdwJ2bPD/h5gaRvsAszmCqKt1Q=
-github.com/livepeer/go-tools v0.3.6/go.mod h1:qs31y68b3PQPmSr8nR8l5WQiIWI623z6pqOccqebjos=
+github.com/livepeer/go-tools v0.3.7 h1:CaiwL7r85EkBd0GUxFyNAp/xMmrjTr/GgIlqoiMtoog=
+github.com/livepeer/go-tools v0.3.7/go.mod h1:qs31y68b3PQPmSr8nR8l5WQiIWI623z6pqOccqebjos=
 github.com/livepeer/joy4 v0.1.1 h1:Tz7gVcmvpG/nfUKHU+XJn6Qke/k32mTWMiH9qB0bhnM=
 github.com/livepeer/joy4 v0.1.1/go.mod h1:xkDdm+akniYxVT9KW1Y2Y7Hso6aW+rZObz3nrA9yTHw=
 github.com/livepeer/livepeer-data v0.8.1 h1:FOlCGbV0ws9hY+F88MZmQhLuvPn5nyl1vuKNWaxCW3c=

handlers/handlers_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/pipeline"
+	"github.com/livepeer/go-tools/drivers"
 	"github.com/stretchr/testify/require"
 )
 
@@ -32,6 +33,7 @@ func TestSuccessfulVODUploadHandler(t *testing.T) {
 	callbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
 	defer callbackServer.Close()
 
+	drivers.Testing = true
 	catalystApiHandlers := CatalystAPIHandlersCollection{VODEngine: pipeline.NewStubCoordinator()}
 	var jsonData = `{
 		"url": "http://localhost/input",

handlers/upload.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/julienschmidt/httprouter"
+	"github.com/livepeer/catalyst-api/clients"
 	"github.com/livepeer/catalyst-api/config"
 	"github.com/livepeer/catalyst-api/errors"
 	"github.com/livepeer/catalyst-api/log"
@@ -257,6 +258,10 @@ func (d *CatalystAPIHandlersCollection) handleUploadVOD(w http.ResponseWriter, r
 		return false, errors.WriteHTTPBadRequest(w, "Invalid request payload", fmt.Errorf("invalid value provided for pipeline strategy: %q", uploadVODRequest.PipelineStrategy))
 	}
 
+	if err = checkWritePermission(requestID, uploadVODRequest.ExternalID, hlsTargetURL, mp4TargetURL, fragMp4TargetURL, clipTargetURL, thumbsTargetURL); err != nil {
+		return false, errors.WriteHTTPBadRequest(w, "Invalid request payload", err)
+	}
+
 	log.Log(requestID, "Received VOD Upload request", "pipeline_strategy", uploadVODRequest.PipelineStrategy, "num_profiles", len(uploadVODRequest.Profiles), "hlsTargetURL", hlsTargetURL)
 
 	// Once we're happy with the request, do the rest of the Segmenting stage asynchronously to allow us to
@@ -310,11 +315,33 @@ func toTargetURL(ol UploadVODRequestOutputLocation, reqID string) (*url.URL, err
 			tURL.Host = reqID
 			log.AddContext(reqID, "w3s-url", tURL.String())
 		}
+
 		return tURL, nil
 	}
 	return nil, nil
 }
 
+func checkWritePermission(reqID, externalID string, urls ...*url.URL) error {
+	// we don't want to re-check the same locations so track them with this map
+	alreadyChecked := make(map[string]bool)
+
+	for _, u := range urls {
+		if u == nil || alreadyChecked[u.String()] {
+			continue
+		}
+
+		urlString := u.String()
+		// check write permission by uploading a file
+		err := clients.UploadToOSURL(u.String(), "metadata.json", strings.NewReader(fmt.Sprintf(`{"external_id": "%s"}`, externalID)), time.Second)
+		if err != nil {
+			log.LogError(reqID, "failed write permission check", err, "url", log.RedactURL(urlString))
+			return fmt.Errorf("failed write permission check for %s", urlString)
+		}
+		alreadyChecked[urlString] = true
+	}
+	return nil
+}
+
 func CheckSourceURLValid(sourceURL string) error {
 	if sourceURL == "" {
 		return fmt.Errorf("empty source URL")

test/features/vod.feature

@@ -25,12 +25,17 @@ Feature: VOD Streaming
     Then I get an HTTP response with code "200"
     And my "successful" vod request metrics get recorded
 
-  Scenario: Submit a bad request to `/api/vod`
-    And I submit to the internal "/api/vod" endpoint with "an invalid upload vod request"
+  Scenario Outline: Submit a bad request to `/api/vod`
+    And I submit to the internal "/api/vod" endpoint with "<payload>"
     And receive a response within "3" seconds
     Then I get an HTTP response with code "400"
     And my "failed" vod request metrics get recorded
 
+    Examples:
+      | payload                                             |
+      | an invalid upload vod request                       |
+      | a valid upload vod request with no write permission |
+
   Scenario Outline: Submit a video asset for ingestion with the FFMPEG / Livepeer pipeline
     When I submit to the internal "/api/vod" endpoint with "<payload>"
     And receive a response within "3" seconds

test/steps/http.go

@@ -114,7 +114,10 @@ func (s *StepContext) postRequest(baseURL, endpoint, payload string, headers map
 	}
 	s.TranscodedOutputDir = destinationDir
 
-	req := DefaultUploadRequest
+	var (
+		req     = DefaultUploadRequest(destinationDir)
+		reqBody string
+	)
 	if strings.HasPrefix(payload, "a valid upload vod request") {
 		req.PipelineStrategy = "fallback_external"
 		req.URL = "file://" + sourceFile.Name()
@@ -130,7 +133,7 @@ func (s *StepContext) postRequest(baseURL, endpoint, payload string, headers map
 				},
 			}
 		}
-		if payload, err = req.ToJSON(); err != nil {
+		if reqBody, err = req.ToJSON(); err != nil {
 			return fmt.Errorf("failed to build upload request JSON: %s", err)
 		}
 	}
@@ -154,7 +157,7 @@ func (s *StepContext) postRequest(baseURL, endpoint, payload string, headers map
 		if strings.Contains(payload, "and thumbnails") {
 			req.OutputLocations[0].Outputs.Thumbnails = "enabled"
 		}
-		if payload, err = req.ToJSON(); err != nil {
+		if reqBody, err = req.ToJSON(); err != nil {
 			return fmt.Errorf("failed to build upload request JSON: %s", err)
 		}
 	}
@@ -174,15 +177,24 @@ func (s *StepContext) postRequest(baseURL, endpoint, payload string, headers map
 		if strings.Contains(payload, "and thumbnails") {
 			req.OutputLocations[0].Outputs.Thumbnails = "enabled"
 		}
-		if payload, err = req.ToJSON(); err != nil {
+		if reqBody, err = req.ToJSON(); err != nil {
+			return fmt.Errorf("failed to build upload request JSON: %s", err)
+		}
+	}
+	if strings.HasSuffix(payload, "with no write permission") {
+		req.OutputLocations[0].URL = "s3+https://u:p@gateway.storjshare.io/foo/bar"
+		if reqBody, err = req.ToJSON(); err != nil {
 			return fmt.Errorf("failed to build upload request JSON: %s", err)
 		}
 	}
 	if payload == "an invalid upload vod request" {
-		payload = "{}"
+		reqBody = "{}"
 	}
 
-	r, err := http.NewRequest(http.MethodPost, baseURL+endpoint, strings.NewReader(payload))
+	if reqBody == "" {
+		reqBody = payload
+	}
+	r, err := http.NewRequest(http.MethodPost, baseURL+endpoint, strings.NewReader(reqBody))
 	r.Header.Set("Authorization", s.authHeaders)
 	r.Header.Set("Content-Type", "application/json")
 	for k, v := range headers {

test/steps/upload_request.go

@@ -29,17 +29,19 @@ type UploadRequest struct {
 	Profiles              []video.EncodedProfile `json:"profiles,omitempty"`
 }
 
-var DefaultUploadRequest = UploadRequest{
-	CallbackURL: "http://localhost:3333/callback/123",
-	OutputLocations: []OutputLocation{
-		{
-			Type: "object_store",
-			URL:  "memory://localhost/output.m3u8",
-			Outputs: Output{
-				HLS: "enabled",
+func DefaultUploadRequest(dest string) UploadRequest {
+	return UploadRequest{
+		CallbackURL: "http://localhost:3333/callback/123",
+		OutputLocations: []OutputLocation{
+			{
+				Type: "object_store",
+				URL:  "file://" + dest,
+				Outputs: Output{
+					HLS: "enabled",
+				},
 			},
 		},
-	},
+	}
 }
 
 func (u UploadRequest) ToJSON() (string, error) {