diff --git a/platform/api/artifact_download_handlers_test.go b/platform/api/artifact_download_handlers_test.go index 095e7b2..0177f28 100644 --- a/platform/api/artifact_download_handlers_test.go +++ b/platform/api/artifact_download_handlers_test.go @@ -175,7 +175,7 @@ func uploadCompletedArtifact(t *testing.T, router http.Handler, sessionToken str Checksum: validator.BytesChecksum(part), Payload: part, } - assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", chunk), http.StatusOK) + assertStatus(t, performArtifactChunkUpload(t, router, chunk), http.StatusOK) } completeRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/complete", dto.ArtifactTransferCompleteRequest{RunEndpointID: "run-local", SessionToken: sessionToken, TransferID: opened.TransferID, ArtifactID: artifactID, Checksum: validator.BytesChecksum(payload), SizeBytes: int64(len(payload))}) assertStatus(t, completeRecorder, http.StatusOK) diff --git a/platform/api/artifact_transfer_handlers_test.go b/platform/api/artifact_transfer_handlers_test.go index 34a432f..3ef47d1 100644 --- a/platform/api/artifact_transfer_handlers_test.go +++ b/platform/api/artifact_transfer_handlers_test.go @@ -1,8 +1,10 @@ package api import ( + "bytes" "net/http" "net/http/httptest" + "strconv" "testing" "browser.local/platform/domain" @@ -23,14 +25,14 @@ func TestArtifactTransferAPIWorkflow(t *testing.T) { t.Fatalf("unexpected open response: %+v", opened) } - chunkRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, 0, 8)) + chunkRecorder := performArtifactChunkUpload(t, router, validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, 0, 8)) assertStatus(t, chunkRecorder, http.StatusOK) chunk := decodeBody[dto.ArtifactChunkUploadResponse](t, chunkRecorder) if !chunk.Accepted || chunk.NextMissingChunkIndex != 1 || len(chunk.ReceivedChunkIndexes) != 1 { t.Fatalf("unexpected chunk response: %+v", chunk) } - duplicateRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, 0, 8)) + duplicateRecorder := performArtifactChunkUpload(t, router, validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, 0, 8)) assertStatus(t, duplicateRecorder, http.StatusOK) duplicate := decodeBody[dto.ArtifactChunkUploadResponse](t, duplicateRecorder) if !duplicate.Duplicate { @@ -48,7 +50,7 @@ func TestArtifactTransferAPIWorkflow(t *testing.T) { assertErrorResponse(t, missingComplete, http.StatusBadRequest, errorCodeValidation) for index := 1; index < opened.TotalChunks; index++ { - partRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, index, 8)) + partRecorder := performArtifactChunkUpload(t, router, validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, index, 8)) assertStatus(t, partRecorder, http.StatusOK) } completeRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/complete", dto.ArtifactTransferCompleteRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, TransferID: opened.TransferID, ArtifactID: "artifact-1", Checksum: validator.BytesChecksum(payload), SizeBytes: int64(len(payload))}) @@ -68,7 +70,7 @@ func TestArtifactTransferAPIErrors(t *testing.T) { badChunk := validArtifactChunkRequest(hello.SessionToken, opened.TransferID, payload, 0, 8) badChunk.Checksum = validator.BytesChecksum([]byte("different")) - badChunkRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", badChunk) + badChunkRecorder := performArtifactChunkUpload(t, router, badChunk) assertErrorResponse(t, badChunkRecorder, http.StatusBadRequest, errorCodeValidation) invalidSession := validArtifactTransferOpenRequest("stale-token", payload, 8) @@ -108,6 +110,23 @@ func performArtifactTransferOpen(t *testing.T, router http.Handler, request dto. return recorder } +func performArtifactChunkUpload(t *testing.T, router http.Handler, request dto.ArtifactChunkUploadRequest) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/api/v1/run/artifacts/chunks", bytes.NewReader(request.Payload)) + req.Header.Set("Content-Type", "application/octet-stream") + req.Header.Set("X-Run-Endpoint", request.RunEndpointID) + req.Header.Set("X-Run-Session-Token", request.SessionToken) + req.Header.Set("X-Artifact-Transfer-Id", request.TransferID) + req.Header.Set("X-Artifact-Id", request.ArtifactID) + req.Header.Set("X-Artifact-Chunk-Index", strconv.Itoa(request.ChunkIndex)) + req.Header.Set("X-Artifact-Offset", strconv.FormatInt(request.Offset, 10)) + req.Header.Set("X-Artifact-Size", strconv.Itoa(request.SizeBytes)) + req.Header.Set("X-Artifact-Checksum", request.Checksum) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, req) + return recorder +} + func validArtifactTransferOpenRequest(sessionToken string, payload []byte, chunkSize int) dto.ArtifactTransferOpenRequest { return dto.ArtifactTransferOpenRequest{ RunEndpointID: "run-local", diff --git a/platform/api/channel_isolation_handlers_test.go b/platform/api/channel_isolation_handlers_test.go index 8f7cd4f..2a9c198 100644 --- a/platform/api/channel_isolation_handlers_test.go +++ b/platform/api/channel_isolation_handlers_test.go @@ -33,7 +33,7 @@ func TestRunChannelAPIInterleavedRequestsMutateIndependentState(t *testing.T) { openRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/open", artifactOpenForChannelIsolation(hello.SessionToken, payload, 8)) assertStatus(t, openRecorder, http.StatusOK) open := decodeBody[dto.ArtifactTransferOpenResponse](t, openRecorder) - firstChunkRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", artifactChunkForChannelIsolation(hello.SessionToken, open.TransferID, payload, 0, 8)) + firstChunkRecorder := performArtifactChunkUpload(t, router, artifactChunkForChannelIsolation(hello.SessionToken, open.TransferID, payload, 0, 8)) assertStatus(t, firstChunkRecorder, http.StatusOK) firstChunk := decodeBody[dto.ArtifactChunkUploadResponse](t, firstChunkRecorder) if !firstChunk.Accepted || firstChunk.NextMissingChunkIndex != 1 { diff --git a/platform/api/run_signature.go b/platform/api/run_signature.go index e7bd586..5c3c3b5 100644 --- a/platform/api/run_signature.go +++ b/platform/api/run_signature.go @@ -14,10 +14,11 @@ import ( ) const ( - runEndpointHeader = "X-Run-Endpoint" - runTimestampHeader = "X-Run-Timestamp" - runNonceHeader = "X-Run-Nonce" - runSignatureHeader = "X-Run-Signature" + runEndpointHeader = "X-Run-Endpoint" + runSessionTokenHeader = "X-Run-Session-Token" + runTimestampHeader = "X-Run-Timestamp" + runNonceHeader = "X-Run-Nonce" + runSignatureHeader = "X-Run-Signature" ) type runRequestEnvelope struct { @@ -37,8 +38,8 @@ func (h *coreHandlers) requireRunSignature(next http.HandlerFunc) http.HandlerFu return } r.Body = io.NopCloser(bytes.NewReader(body)) - var envelope runRequestEnvelope - if err := json.Unmarshal(body, &envelope); err != nil { + envelope, ok := signedRunRequestEnvelope(r, body) + if !ok { next(w, r) return } @@ -65,3 +66,19 @@ func (h *coreHandlers) requireRunSignature(next http.HandlerFunc) http.HandlerFu next(w, r) } } + +func signedRunRequestEnvelope(r *http.Request, body []byte) (runRequestEnvelope, bool) { + if isOctetStream(r.Header.Get("Content-Type")) { + return runRequestEnvelope{RunEndpointID: strings.TrimSpace(r.Header.Get(runEndpointHeader)), SessionToken: strings.TrimSpace(r.Header.Get(runSessionTokenHeader))}, true + } + var envelope runRequestEnvelope + if err := json.Unmarshal(body, &envelope); err != nil { + return runRequestEnvelope{}, false + } + return envelope, true +} + +func isOctetStream(contentType string) bool { + mediaType := strings.ToLower(strings.TrimSpace(strings.Split(contentType, ";")[0])) + return mediaType == "application/octet-stream" +} diff --git a/platform/dto/artifact_transfer.go b/platform/dto/artifact_transfer.go index 11a4329..aa2a521 100644 --- a/platform/dto/artifact_transfer.go +++ b/platform/dto/artifact_transfer.go @@ -42,7 +42,7 @@ type ArtifactChunkUploadRequest struct { Offset int64 `json:"offset"` SizeBytes int `json:"sizeBytes"` Checksum string `json:"checksum"` - Payload []byte `json:"payload"` + Payload []byte `json:"-"` } type ArtifactChunkUploadResponse struct { diff --git a/platform/service/artifact_body_store.go b/platform/service/artifact_body_store.go index 8bc9782..a9e6f43 100644 --- a/platform/service/artifact_body_store.go +++ b/platform/service/artifact_body_store.go @@ -1,6 +1,7 @@ package service import ( + "bytes" "crypto/sha256" "encoding/hex" "encoding/json" @@ -24,6 +25,7 @@ type ArtifactBodyStore interface { PutPayload(string, []byte) error GetPayload(string) ([]byte, error) ReadPayloadRange(string, int64, int) ([]byte, error) + OpenPayloadRange(string, int64, int64) (io.ReadCloser, error) CommitTransferPayload(domain.ArtifactTransferSession) error } @@ -89,6 +91,17 @@ func (store *MemoryArtifactBodyStore) ReadPayloadRange(artifactID string, offset return domain.CopyBytes(payload[int(offset) : int(offset)+length]), nil } +func (store *MemoryArtifactBodyStore) OpenPayloadRange(artifactID string, offset int64, length int64) (io.ReadCloser, error) { + if length < 0 || length > int64(int(^uint(0)>>1)) { + return nil, fmt.Errorf("artifact range is invalid") + } + payload, err := store.ReadPayloadRange(artifactID, offset, int(length)) + if err != nil { + return nil, err + } + return io.NopCloser(bytes.NewReader(payload)), nil +} + func (store *MemoryArtifactBodyStore) CommitTransferPayload(session domain.ArtifactTransferSession) error { store.mu.Lock() defer store.mu.Unlock() @@ -243,6 +256,40 @@ func (store *FileArtifactBodyStore) ReadPayloadRange(artifactID string, offset i return payload, nil } +func (store *FileArtifactBodyStore) OpenPayloadRange(artifactID string, offset int64, length int64) (io.ReadCloser, error) { + if offset < 0 || length < 0 { + return nil, fmt.Errorf("artifact range is invalid") + } + store.mu.Lock() + defer store.mu.Unlock() + file, err := os.Open(store.payloadPath(artifactID)) + if errors.Is(err, os.ErrNotExist) { + return nil, repo.ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("open artifact payload: %w", err) + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return nil, fmt.Errorf("stat artifact payload: %w", err) + } + if offset > info.Size() || length > info.Size()-offset { + _ = file.Close() + return nil, fmt.Errorf("artifact range is invalid") + } + return sectionReadCloser{SectionReader: io.NewSectionReader(file, offset, length), file: file}, nil +} + +type sectionReadCloser struct { + *io.SectionReader + file *os.File +} + +func (reader sectionReadCloser) Close() error { + return reader.file.Close() +} + func (store *FileArtifactBodyStore) CommitTransferPayload(session domain.ArtifactTransferSession) error { store.mu.Lock() defer store.mu.Unlock() diff --git a/platform/service/artifact_download.go b/platform/service/artifact_download.go index c1f21ab..71b1fe8 100644 --- a/platform/service/artifact_download.go +++ b/platform/service/artifact_download.go @@ -3,6 +3,7 @@ package service import ( "errors" "fmt" + "io" "net/url" "sort" "strings" @@ -16,6 +17,11 @@ import ( const artifactDownloadStorageBehavior = "platform-durable-artifact-store" const artifactPayloadCacheLimit = 4 * 1024 * 1024 +type ArtifactContentStream struct { + Content domain.ArtifactContent + Body io.ReadCloser +} + func (svc *CoreService) GetArtifactForSession(sessionID string, artifactID string) (domain.Artifact, error) { artifact, err := svc.store.Artifacts().Get(strings.TrimSpace(artifactID)) if err != nil { @@ -74,22 +80,11 @@ func (svc *CoreService) ReadArtifactContentForSession(sessionID string, request if artifact.State != domain.ArtifactStateAvailable { return domain.ArtifactContent{}, validationError("artifact must be available before download") } - limit := request.Limit - if limit == 0 { - if request.Offset == 0 { - limit = int(artifact.SizeBytes) - } else { - limit = validator.MaxArtifactDownloadBytes - } + offset, length, partial, err := artifactContentBounds(artifact, request) + if err != nil { + return domain.ArtifactContent{}, err } - if request.Offset > artifact.SizeBytes { - return domain.ArtifactContent{}, validationError("artifact range exceeds metadata") - } - remaining := artifact.SizeBytes - request.Offset - if int64(limit) > remaining { - limit = int(remaining) - } - payload, err := svc.artifactStore.ReadPayloadRange(artifact.ID, request.Offset, limit) + payload, err := svc.artifactStore.ReadPayloadRange(artifact.ID, offset, int(length)) if err != nil { return domain.ArtifactContent{}, err } @@ -99,12 +94,12 @@ func (svc *CoreService) ReadArtifactContentForSession(sessionID string, request ArtifactID: artifact.ID, Filename: filename, ContentType: contentType, - Offset: request.Offset, + Offset: offset, SizeBytes: int64(len(part)), TotalSizeBytes: artifact.SizeBytes, Checksum: artifact.Checksum, ContentChecksum: validator.BytesChecksum(part), - Partial: request.Offset != 0 || int64(len(part)) != artifact.SizeBytes, + Partial: partial, RangeSupported: true, Payload: part, StorageBehavior: artifactDownloadStorageBehavior, @@ -116,6 +111,66 @@ func (svc *CoreService) ReadArtifactContentForSession(sessionID string, request return domain.CopyArtifactContent(content), nil } +func (svc *CoreService) OpenArtifactContentStreamForSession(sessionID string, request domain.ArtifactContentRequest) (ArtifactContentStream, error) { + request = domain.CopyArtifactContentRequest(request) + if err := validator.ValidateArtifactContentRequest(request); err != nil { + return ArtifactContentStream{}, err + } + artifact, err := svc.GetArtifactForSession(sessionID, request.ArtifactID) + if err != nil { + return ArtifactContentStream{}, err + } + if artifact.State != domain.ArtifactStateAvailable { + return ArtifactContentStream{}, validationError("artifact must be available before download") + } + offset, length, partial, err := artifactContentBounds(artifact, request) + if err != nil { + return ArtifactContentStream{}, err + } + contentChecksum := artifact.Checksum + if partial { + payload, err := svc.artifactStore.ReadPayloadRange(artifact.ID, offset, int(length)) + if err != nil { + return ArtifactContentStream{}, err + } + contentChecksum = validator.BytesChecksum(payload) + } + reader, err := svc.artifactStore.OpenPayloadRange(artifact.ID, offset, length) + if err != nil { + return ArtifactContentStream{}, err + } + filename, contentType := svc.artifactDownloadPresentation(artifact) + content := domain.ArtifactContent{ArtifactID: artifact.ID, Filename: filename, ContentType: contentType, Offset: offset, SizeBytes: length, TotalSizeBytes: artifact.SizeBytes, Checksum: artifact.Checksum, ContentChecksum: contentChecksum, Partial: partial, RangeSupported: true, StorageBehavior: artifactDownloadStorageBehavior, ServedAt: svc.now()} + if err := validator.ValidateArtifactContent(content); err != nil { + _ = reader.Close() + return ArtifactContentStream{}, err + } + return ArtifactContentStream{Content: domain.CopyArtifactContent(content), Body: reader}, nil +} + +func artifactContentBounds(artifact domain.Artifact, request domain.ArtifactContentRequest) (int64, int64, bool, error) { + if request.Offset > artifact.SizeBytes { + return 0, 0, false, validationError("artifact range exceeds metadata") + } + length := int64(request.Limit) + if request.Limit == 0 { + if request.Offset == 0 { + length = artifact.SizeBytes + } else { + length = int64(validator.MaxArtifactDownloadBytes) + } + } + remaining := artifact.SizeBytes - request.Offset + if length > remaining { + length = remaining + } + if length <= 0 { + return 0, 0, false, validationError("artifact range is empty") + } + partial := request.Offset != 0 || length != artifact.SizeBytes + return request.Offset, length, partial, nil +} + func (svc *CoreService) authorizeArtifactAccess(sessionID string, artifact domain.Artifact) error { user, err := svc.GetCurrentUser(sessionID) if err != nil { diff --git a/platform/validator/artifact_download.go b/platform/validator/artifact_download.go index a386d90..6b67d65 100644 --- a/platform/validator/artifact_download.go +++ b/platform/validator/artifact_download.go @@ -101,17 +101,22 @@ func ValidateArtifactContent(content domain.ArtifactContent) error { if content.SizeBytes > maxContentBytes { violations = append(violations, fmt.Sprintf("sizeBytes must not exceed %d", maxContentBytes)) } - if int64(len(content.Payload)) != content.SizeBytes { - violations = append(violations, "payload size must match sizeBytes") - } if content.Offset+content.SizeBytes > content.TotalSizeBytes { violations = append(violations, "range exceeds artifact size") } if content.Checksum != "" && !validSHA256Checksum(content.Checksum) { violations = append(violations, "checksum must be sha256:") } - if content.ContentChecksum != "" && content.ContentChecksum != BytesChecksum(content.Payload) { - violations = append(violations, "contentChecksum does not match payload") + if content.ContentChecksum != "" && !validSHA256Checksum(content.ContentChecksum) { + violations = append(violations, "contentChecksum must be sha256:") + } + if content.Payload != nil { + if int64(len(content.Payload)) != content.SizeBytes { + violations = append(violations, "payload size must match sizeBytes") + } + if content.ContentChecksum != "" && content.ContentChecksum != BytesChecksum(content.Payload) { + violations = append(violations, "contentChecksum does not match payload") + } } for _, value := range []fieldString{ {field: "filename", value: content.Filename},