From 557ef34a4426671ceb060afd92285760ec1f8571 Mon Sep 17 00:00:00 2001 From: Javier Rodriguez Date: Tue, 25 Aug 2026 08:23:56 +0200 Subject: [PATCH] feat(artifact-cas): verify uploaded content against the declared digest The Artifact CAS keys stored artifacts by their client-declared SHA256. This change adds end-to-end content-integrity verification to the upload path so the canonical key is only ever written from bytes CAS has confirmed hash to that digest, across all object-store backends (S3, S3 access point, Azure Blob), in addition to the OCI backend's existing layer-digest check. The CAS service streams each upload to a temporary file on a local staging volume, computes its SHA256 while receiving, and verifies it against the declared digest before handing the verified file to the backend. A mismatch is rejected with InvalidArgument and nothing is written to the backend. The upload path is consolidated into a single spill-then-upload flow shared by all backends, replacing the per-backend streaming branch and the StreamingUploader interface. The staging directory is configured through staging_dir and is required: the CAS refuses to start without one, and proves it is writable before serving traffic, so a misconfiguration surfaces at boot rather than as a healthy service that fails every upload. The Helm chart mounts a dedicated emptyDir for it. Each staging file is unlinked as soon as it is created and the upload proceeds through the open descriptor, so the kernel reclaims the space whenever the process exits, including when it is killed mid-upload, and the volume cannot accumulate partial artifacts. Assisted-by: Claude Code Signed-off-by: Javier Rodriguez Chainloop-Trace-Sessions: 5d0fc605-25b0-4bb3-900c-2c2a8fb4922f --- app/artifact-cas/cmd/main.go | 46 +++ app/artifact-cas/cmd/main_test.go | 88 +++++ app/artifact-cas/cmd/wire.go | 3 +- app/artifact-cas/cmd/wire_gen.go | 6 +- app/artifact-cas/configs/config.devel.yaml | 5 + app/artifact-cas/internal/conf/conf.pb.go | 21 +- app/artifact-cas/internal/conf/conf.proto | 6 + .../internal/service/auditor_test.go | 3 + .../internal/service/bytestream.go | 310 ++++++----------- .../service/bytestream_download_test.go | 10 +- .../internal/service/bytestream_e2e_test.go | 304 ++++++++++++++++ .../service/bytestream_streaming_test.go | 326 ++++++++++++------ .../internal/service/bytestream_test.go | 112 +++--- app/artifact-cas/internal/service/service.go | 15 + deployment/chainloop/Chart.yaml | 2 +- deployment/chainloop/templates/_helpers.tpl | 10 + .../chainloop/templates/cas/configmap.yaml | 3 + .../chainloop/templates/cas/deployment.yaml | 9 + deployment/chainloop/values.yaml | 16 +- pkg/blobmanager/azureblob/backend.go | 33 +- pkg/blobmanager/azureblob/backend_test.go | 34 -- pkg/blobmanager/backend.go | 16 - pkg/blobmanager/oci/backend_test.go | 12 - pkg/blobmanager/s3/backend.go | 18 +- pkg/blobmanager/s3/backend_test.go | 26 -- pkg/blobmanager/s3accesspoint/backend.go | 11 +- pkg/blobmanager/s3accesspoint/backend_test.go | 11 - 27 files changed, 949 insertions(+), 507 deletions(-) create mode 100644 app/artifact-cas/cmd/main_test.go create mode 100644 app/artifact-cas/internal/service/bytestream_e2e_test.go delete mode 100644 pkg/blobmanager/azureblob/backend_test.go diff --git a/app/artifact-cas/cmd/main.go b/app/artifact-cas/cmd/main.go index b3986301f..5e64b440d 100644 --- a/app/artifact-cas/cmd/main.go +++ b/app/artifact-cas/cmd/main.go @@ -16,7 +16,9 @@ package main import ( + "errors" "flag" + "fmt" "os" "time" @@ -121,6 +123,12 @@ func main() { _ = logger.Log(log.LevelInfo, "msg", "starting artifact-cas service", "version", Version) + // Ensure the upload staging directory is configured and writable before we + // accept any traffic: without it no upload can be verified. + if err := prepareStagingDir(&bc); err != nil { + panic(err) + } + flush, err := initSentry(&bc, logger) defer flush() if err != nil { @@ -152,6 +160,44 @@ func newProtoValidator() (protovalidate.Validator, error) { return protovalidate.New() } +// prepareStagingDir creates the upload staging directory and proves it is +// writable. It must be the same directory the service is configured with (see +// serviceOpts / conf.staging_dir). +// +// staging_dir is required: uploads are verified by spilling them here first, so +// a missing or unwritable directory means no upload can succeed. Failing at +// startup surfaces that immediately, rather than letting the service report +// healthy and reject every upload. The CAS container runs with a read-only root +// filesystem and /tmp is a read-only secret mount, so the configured directory is +// the only place uploads can be staged. +// +// The directory stays clean on its own: each upload removes its staging file on +// every exit path, and the emptyDir backing it is cleared by Kubernetes when the +// Pod is removed from the node. +func prepareStagingDir(bc *conf.Bootstrap) error { + dir := bc.GetStagingDir() + if dir == "" { + return errors.New("staging_dir is required: it must point at a writable, pod-local directory") + } + + if err := os.MkdirAll(dir, 0o700); err != nil { + return fmt.Errorf("creating staging dir %q: %w", dir, err) + } + + // MkdirAll succeeds on a directory that already exists but cannot be written + // to, so probe it rather than discovering the problem on the first upload. + probe, err := os.CreateTemp(dir, ".writable-probe-*") + if err != nil { + return fmt.Errorf("staging dir %q is not writable: %w", dir, err) + } + _ = probe.Close() + if err := os.Remove(probe.Name()); err != nil { + return fmt.Errorf("removing staging dir probe file: %w", err) + } + + return nil +} + func initSentry(c *conf.Bootstrap, logger log.Logger) (cleanupFunc func(), err error) { cleanupFunc = func() { sentry.Flush(2 * time.Second) diff --git a/app/artifact-cas/cmd/main_test.go b/app/artifact-cas/cmd/main_test.go new file mode 100644 index 000000000..c97004007 --- /dev/null +++ b/app/artifact-cas/cmd/main_test.go @@ -0,0 +1,88 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "os" + "path/filepath" + "testing" + + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" + "github.com/stretchr/testify/require" +) + +// TestPrepareStagingDir: the staging directory is a hard startup requirement. +// Every way it can be unusable must fail the boot, because a CAS that starts +// without one reports healthy and then rejects every upload. +func TestPrepareStagingDir(t *testing.T) { + testCases := []struct { + name string + // dir builds the staging_dir value for the case; t.TempDir() gives each + // case its own scratch space. + dir func(t *testing.T) string + expectErr string + }{ + { + name: "unconfigured is rejected rather than defaulted", + dir: func(*testing.T) string { return "" }, + expectErr: "staging_dir is required", + }, + { + name: "an existing writable directory is accepted", + dir: func(t *testing.T) string { return t.TempDir() }, + }, + { + name: "a missing directory is created", + dir: func(t *testing.T) string { + return filepath.Join(t.TempDir(), "nested", "staging") + }, + }, + { + name: "an existing but unwritable directory is rejected", + dir: func(t *testing.T) string { + dir := filepath.Join(t.TempDir(), "read-only") + require.NoError(t, os.Mkdir(dir, 0o500)) + t.Cleanup(func() { _ = os.Chmod(dir, 0o700) }) + return dir + }, + expectErr: "not writable", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + if os.Geteuid() == 0 && tc.expectErr == "not writable" { + t.Skip("root ignores directory permissions") + } + + dir := tc.dir(t) + err := prepareStagingDir(&conf.Bootstrap{StagingDir: dir}) + + if tc.expectErr != "" { + require.ErrorContains(t, err, tc.expectErr) + return + } + + require.NoError(t, err) + require.DirExists(t, dir) + + // The writability probe must not survive the check. + entries, err := os.ReadDir(dir) + require.NoError(t, err) + require.Empty(t, entries, "the probe file must be removed") + }) + } +} diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go index da2a3735e..a3866888e 100644 --- a/app/artifact-cas/cmd/wire.go +++ b/app/artifact-cas/cmd/wire.go @@ -51,10 +51,11 @@ func wireApp(*conf.Bootstrap, *conf.Server, *conf.Auth, credentials.Reader, log. ) } -func serviceOpts(l log.Logger, audit *service.AuditDispatcher) []service.NewOpt { +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { return []service.NewOpt{ service.WithLogger(l), service.WithAuditDispatcher(audit), + service.WithStagingDir(bc.GetStagingDir()), } } diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index 6ae068607..4b48eef04 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -37,7 +37,7 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth return nil, nil, err } auditDispatcher := service.NewAuditDispatcher(auditLogPublisher, logger) - v := serviceOpts(logger, auditDispatcher) + v := serviceOpts(logger, auditDispatcher, bootstrap) byteStreamService := service.NewByteStreamService(providers, v...) resourceService := service.NewResourceService(providers, v...) validator, err := newProtoValidator() @@ -75,8 +75,8 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth // wire.go: -func serviceOpts(l log.Logger, audit *service.AuditDispatcher) []service.NewOpt { - return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit)} +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { + return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir())} } // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset diff --git a/app/artifact-cas/configs/config.devel.yaml b/app/artifact-cas/configs/config.devel.yaml index 63c53c6cd..50c84adb6 100644 --- a/app/artifact-cas/configs/config.devel.yaml +++ b/app/artifact-cas/configs/config.devel.yaml @@ -39,3 +39,8 @@ observability: auth: public_key_path: ${PUBLIC_KEY_PATH:../../devel/devkeys/cas.pub} + +# Local directory where uploads are staged and verified against the declared +# digest before being sent to the backend. Required: there is no fallback. +# In Kubernetes this points at a dedicated per-pod emptyDir. +staging_dir: ${STAGING_DIR:/tmp/chainloop-cas-staging} diff --git a/app/artifact-cas/internal/conf/conf.pb.go b/app/artifact-cas/internal/conf/conf.pb.go index e724c6098..c364830d7 100644 --- a/app/artifact-cas/internal/conf/conf.pb.go +++ b/app/artifact-cas/internal/conf/conf.pb.go @@ -46,7 +46,13 @@ type Bootstrap struct { CredentialsService *v1.Credentials `protobuf:"bytes,4,opt,name=credentials_service,json=credentialsService,proto3" json:"credentials_service,omitempty"` // Optional NATS server configuration to publish audit events to the // control-plane-owned stream. When unset, event publishing is disabled. - NatsServer *Bootstrap_NatsServer `protobuf:"bytes,5,opt,name=nats_server,json=natsServer,proto3" json:"nats_server,omitempty"` + NatsServer *Bootstrap_NatsServer `protobuf:"bytes,5,opt,name=nats_server,json=natsServer,proto3" json:"nats_server,omitempty"` + // Local directory where uploads (and, later, downloads) are staged on disk + // and verified against the declared digest before reaching the backend. It + // must be a writable volume; in production a dedicated emptyDir is mounted + // here (NOT tmpfs/RAM, and NOT the /tmp secret mount). When unset the service + // falls back to the OS temp dir, which is only appropriate for local dev. + StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -116,6 +122,13 @@ func (x *Bootstrap) GetNatsServer() *Bootstrap_NatsServer { return nil } +func (x *Bootstrap) GetStagingDir() string { + if x != nil { + return x.StagingDir + } + return "" +} + type Server struct { state protoimpl.MessageState `protogen:"open.v1"` // Regular HTTP endpoint @@ -729,14 +742,16 @@ var File_conf_proto protoreflect.FileDescriptor const file_conf_proto_rawDesc = "" + "\n" + "\n" + - "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xb9\x05\n" + + "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xda\x05\n" + "\tBootstrap\x12\x1f\n" + "\x06server\x18\x01 \x01(\v2\a.ServerR\x06server\x12\x19\n" + "\x04auth\x18\x02 \x01(\v2\x05.AuthR\x04auth\x12>\n" + "\robservability\x18\x03 \x01(\v2\x18.Bootstrap.ObservabilityR\robservability\x12L\n" + "\x13credentials_service\x18\x04 \x01(\v2\x1b.credentials.v1.CredentialsR\x12credentialsService\x126\n" + "\vnats_server\x18\x05 \x01(\v2\x15.Bootstrap.NatsServerR\n" + - "natsServer\x1aH\n" + + "natsServer\x12\x1f\n" + + "\vstaging_dir\x18\x06 \x01(\tR\n" + + "stagingDir\x1aH\n" + "\n" + "NatsServer\x12\x10\n" + "\x03uri\x18\x01 \x01(\tR\x03uri\x12\x16\n" + diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto index 9905e1c03..5c14b4227 100644 --- a/app/artifact-cas/internal/conf/conf.proto +++ b/app/artifact-cas/internal/conf/conf.proto @@ -28,6 +28,12 @@ message Bootstrap { // Optional NATS server configuration to publish audit events to the // control-plane-owned stream. When unset, event publishing is disabled. NatsServer nats_server = 5; + // Local directory where uploads (and, later, downloads) are staged on disk + // and verified against the declared digest before reaching the backend. It + // must be a writable volume; in production a dedicated emptyDir is mounted + // here (NOT tmpfs/RAM, and NOT the /tmp secret mount). When unset the service + // falls back to the OS temp dir, which is only appropriate for local dev. + string staging_dir = 6; message NatsServer { // NATS server URI, e.g. "nats://localhost:4222" diff --git a/app/artifact-cas/internal/service/auditor_test.go b/app/artifact-cas/internal/service/auditor_test.go index 267ac5e29..0af3eca6b 100644 --- a/app/artifact-cas/internal/service/auditor_test.go +++ b/app/artifact-cas/internal/service/auditor_test.go @@ -71,6 +71,9 @@ func testUploadedEntry() auditor.LogEntry { const testOrgID = "1089bb36-e27b-428b-8009-d015c8737c54" +// testStoredSecretID is the credentials handle the fake JWT claims carry. +const testStoredSecretID = "secret-id" + func TestAuditDispatcherDispatch(t *testing.T) { tests := []struct { name string diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 86c588141..8ea5f2b51 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -25,6 +25,7 @@ import ( "fmt" "hash" "io" + "os" "errors" @@ -116,30 +117,30 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro s.log.Infow("msg", "artifact does not exist, uploading", "digest", req.resource.Digest, "name", req.resource.FileName) - // Streaming-capable backends (object stores such as S3/Azure) are fed - // directly from the client stream through an io.Pipe, so CAS memory stays - // bounded by the chunk/pipe size regardless of artifact size (PFM-6923). - // The OCI backend, whose push path needs the whole layer content up front, - // does not advertise streaming and keeps the fully-buffered path. - var committedSize int64 - if su, ok := storageBackend.(backend.StreamingUploader); ok && su.SupportsStreaming() { - committedSize, err = s.streamUpload(ctx, stream, storageBackend, req, info.MaxBytes) - } else { - committedSize, err = s.bufferedUpload(ctx, stream, storageBackend, req, info.MaxBytes) - } - - // Classify the outcome. The error may come from two distinct stages, which - // must be treated differently: + // Spill the upload to local disk, verify its SHA256 against the declared + // digest, and only then hand the verified file to the backend. The canonical + // key can therefore never hold content that does not hash to its digest, and + // CAS memory stays bounded because the artifact lives on disk. + committedSize, err := s.spillVerifyUpload(ctx, stream, storageBackend, req, info.MaxBytes) + + // Classify the outcome. The error may come from several distinct stages, + // which must be treated differently: + // - A digest mismatch (digestMismatchError) is the client's fault: the + // bytes do not hash to the key they declared, so the request is invalid + // and no bytes were ever sent to the backend. // - A backend Upload failure (backendUploadError) is always masked as an // internal error. It must NOT be interpreted as a client disconnect even // when it wraps a network reset/cancellation originating backend-side — // doing so would falsely report success and silently drop the artifact. - // - A stream-read (feed) error is classified: a client disconnect is not a + // - A stream-read (spill) error is classified: a client disconnect is not a // failure, an exceeded size cap maps to ResourceExhausted, anything else - // is masked. + // (e.g. a staging-disk write failure) is masked. if err != nil { - var backendErr *backendUploadError - if errors.As(err, &backendErr) { + if mismatch, ok := errors.AsType[*digestMismatchError](err); ok { + s.log.Infow("msg", "upload rejected: digest mismatch", "digest", req.resource.Digest, "name", req.resource.FileName, "got", mismatch.got) + return status.Error(codes.InvalidArgument, err.Error()) + } + if backendErr, ok := errors.AsType[*backendUploadError](err); ok { return sl.LogAndMaskErr(backendErr.err, s.log) } if isClientDisconnect(err) { @@ -165,91 +166,97 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return stream.SendAndClose(&bytestream.WriteResponse{CommittedSize: committedSize}) } -// bufferedUpload accumulates the whole artifact in memory before handing it to -// the backend. This is required by the OCI backend: its push implementation -// does not support streaming/chunked uploads for uncompressed layers (we can not -// use stream.Layer since it only supports compressed layers, and we want to -// store raw data with custom mimetypes), so it needs the full content up front. -// https://github.com/google/go-containerregistry/blob/main/pkg/v1/stream/README.md -// It returns the total number of bytes committed to the backend. Feed errors are -// returned unwrapped (classified by the caller); backend Upload failures are -// wrapped in backendUploadError so the caller always masks them. -func (s *ByteStreamService) bufferedUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) { - // Create a buffer that will be filled in the background before sending its content to the backend - buffer := newStreamReader(maxBytes) - // Add data from the first request - if err := buffer.Write(req.GetData()); err != nil { - return 0, err +// spillVerifyUpload stages an upload on local disk, verifies its digest, and +// stores it. The client's bytes are streamed into a temporary file and hashed in +// the same pass; the file is handed to the backend only once the hash matches the +// client-declared digest, so unverified bytes never reach the canonical key. +// Staging on disk keeps this service's memory use independent of artifact size. +// +// It returns the number of bytes committed. A digest mismatch is returned as a +// *digestMismatchError; a backend Upload failure as a *backendUploadError; spill +// errors (client disconnect, exceeded size cap, staging-disk write failure) are +// returned unwrapped for the caller to classify. +func (s *ByteStreamService) spillVerifyUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) { + // os.CreateTemp falls back to the OS temp dir when given an empty path, which + // on the CAS container is a read-only secret mount. Refuse instead: an + // unconfigured staging dir is a deployment error, not something to paper over. + if s.stagingDir == "" { + return 0, errors.New("no staging directory configured") + } + + f, err := os.CreateTemp(s.stagingDir, stagingFilePrefix+"*") + if err != nil { + return 0, fmt.Errorf("creating staging file: %w", err) + } + + // Drop the directory entry straight away and keep working through the open + // file descriptor. The staged content stays fully readable and seekable, but + // it is now owned by this process rather than by the filesystem: the kernel + // releases the space when the descriptor goes away, including when the + // process is killed outright. + // + // This is what keeps the staging volume bounded. A deferred remove only runs + // when the handler returns, so a SIGKILL or an OOM kill mid-upload would + // strand a partial artifact on the volume, and nothing would ever reclaim it: + // the emptyDir backing it outlives container restarts and is cleared only + // when the Pod is removed from the node. Unlinking up front means an + // interrupted upload cannot leave anything behind, whatever kills us, so no + // sweep or reaper is needed to keep the volume from filling up. + if err := os.Remove(f.Name()); err != nil { + // Closing drops our handle but leaves the file itself in the staging + // directory, so try removing it once more rather than abandoning it there. + _ = f.Close() + if rmErr := os.Remove(f.Name()); rmErr != nil && !errors.Is(rmErr, os.ErrNotExist) { + s.log.Warnw("msg", "staging file left on disk", "path", f.Name(), "error", rmErr.Error()) + } + return 0, fmt.Errorf("unlinking staging file: %w", err) } - // Start a goroutine that will fill the buffer in the background - go bufferStream(ctx, stream, buffer, s.log) + // Closing is now the whole cleanup: it drops the last reference to the + // unlinked inode and frees the space, on every exit path. + defer func() { + if err := f.Close(); err != nil { + s.log.Warnw("msg", "failed to close staging file", "error", err.Error()) + } + }() - // Block until the buffer has been filled or the upload process has been canceled - if err := <-buffer.errorChan; err != nil { + // Tee the stream into the file and a SHA256 hasher in one pass. + hasher := sha256.New() + size, err := spillStream(ctx, stream, io.MultiWriter(f, hasher), req.GetData(), maxBytes, s.log, req.resource.Digest) + if err != nil { return 0, err } - s.log.Infow("msg", "artifact received, uploading now to backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", buffer.size) - if err := storageBackend.Upload(ctx, buffer, req.resource); err != nil { - return 0, &backendUploadError{err} + // Fail closed: if the streamed bytes do not hash to the declared digest, + // reject the upload and send nothing to the backend. + if got := hex.EncodeToString(hasher.Sum(nil)); got != req.resource.Digest { + return 0, &digestMismatchError{got: got, want: req.resource.Digest} } - return buffer.size, nil -} - -// streamUpload pipes the client stream straight into the backend's Upload -// without buffering the whole artifact in memory. A background goroutine feeds -// received chunks into an io.Pipe while Upload consumes the other end, so the -// two run concurrently and peak memory stays bounded (PFM-6923). It returns the -// total number of bytes committed to the backend. -func (s *ByteStreamService) streamUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) { - pr, pw := io.Pipe() - - var ( - uploadedSize int64 - feedErr error - ) - done := make(chan struct{}) - go func() { - defer close(done) - uploadedSize, feedErr = feedPipe(ctx, stream, pw, req.GetData(), maxBytes, s.log, req.resource.Digest) - // Closing with feedErr signals EOF to the reader when nil, or propagates - // the failure so Upload stops reading. - _ = pw.CloseWithError(feedErr) - }() - - uploadErr := storageBackend.Upload(ctx, streamingReader{pr}, req.resource) - // If Upload returned without draining the pipe (a backend failure, or a - // backend that reports success without reading to EOF), the feeding goroutine - // may still be blocked on Write; closing the read end unblocks it. Then wait - // for it so uploadedSize/feedErr are safe to read. - _ = pr.CloseWithError(uploadErr) - <-done - - // errPipeConsumerGone means the feed only failed because the reader (Upload) - // stopped consuming — a consequence of the upload outcome, not a genuine - // stream-read failure, so the backend's own result is authoritative. - if errors.Is(feedErr, errPipeConsumerGone) { - feedErr = nil + // Rewind so the backend reads from the start. A seekable body also lets the + // AWS SDK learn the exact length and take its zero-copy SectionReader fast + // path instead of buffering parts in memory. + if _, err := f.Seek(0, io.SeekStart); err != nil { + return 0, fmt.Errorf("rewinding staging file: %w", err) } - // A genuine feed-side error (client disconnect, exceeded size cap, stream - // read failure) is the precise signal and takes precedence: when it occurs it - // is what induced the backend error through the pipe. Returned unwrapped so - // the caller classifies it (disconnect / ResourceExhausted / mask). - if feedErr != nil { - return 0, feedErr - } - // A backend failure is wrapped so the caller always masks it, never mistaking - // a backend-side reset/cancellation for a client disconnect. - if uploadErr != nil { - return 0, &backendUploadError{uploadErr} + s.log.Infow("msg", "artifact verified, uploading now to backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", size) + // IMPORTANT: hand the *os.File to Upload unwrapped. Wrapping it (io.TeeReader, + // io.LimitReader, a progress reader) hides io.ReaderAt/io.Seeker and silently + // forces the object-store SDK back onto in-memory multipart buffering. + if err := storageBackend.Upload(ctx, f, req.resource); err != nil { + return 0, &backendUploadError{err} } - return uploadedSize, nil + return size, nil } +// stagingFilePrefix names the per-upload temporary files. They are unlinked as +// soon as they are created, so the name never shows up in a directory listing; +// it identifies them in the process's open descriptors, where an in-flight +// upload appears as "/cas-upload-NNN (deleted)". +const stagingFilePrefix = "cas-upload-" + // backendUploadError marks a failure returned by the storage backend's Upload, // as opposed to an error reading the client stream. Backend failures are always // masked as internal errors and are never interpreted as a client disconnect or @@ -259,28 +266,21 @@ type backendUploadError struct{ err error } func (e *backendUploadError) Error() string { return e.err.Error() } func (e *backendUploadError) Unwrap() error { return e.err } -// errPipeConsumerGone is returned by feedPipe when a write to the pipe fails, -// which only happens once the reader (the backend Upload) has stopped consuming -// — because Upload returned and streamUpload closed the read end, or because it -// failed. It is not a genuine stream-read failure; streamUpload defers to the -// backend's own error in that case. -var errPipeConsumerGone = errors.New("pipe consumer stopped reading") - -// streamingReader wraps the upload pipe reader with a stable string form. The -// pipe is written to concurrently while the backend reads it; exposing the bare -// *io.PipeReader lets a reflective consumer (a structured logger, a test's mock -// matcher, etc.) walk the pipe's internal state and race with the writer. The -// wrapper keeps io.Reader behaviour while presenting an opaque identity to fmt. -type streamingReader struct { - io.Reader -} +// digestMismatchError marks an upload whose streamed bytes do not hash to the +// client-declared digest. It is surfaced to the client as InvalidArgument: the +// request is malformed (the declared key does not describe the content), and no +// bytes are ever written to the backend. +type digestMismatchError struct{ got, want string } -func (streamingReader) String() string { return "cas-streaming-upload" } +func (e *digestMismatchError) Error() string { + return fmt.Sprintf("uploaded content does not match the declared digest: got=%s, want=%s", e.got, e.want) +} -// feedPipe forwards the artifact from the client stream into pw, enforcing the -// max upload size as it goes. firstData is the payload already read from the -// first request. It returns the total number of bytes forwarded. -func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw *io.PipeWriter, firstData []byte, maxSize int64, log *log.Helper, digest string) (int64, error) { +// spillStream forwards the artifact from the client stream into w (the staging +// file tee'd into a SHA256 hasher), enforcing the max upload size as it goes. +// firstData is the payload already read from the first request. It returns the +// total number of bytes written. It reads the client stream straight into w. +func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, w io.Writer, firstData []byte, maxSize int64, log *log.Helper, digest string) (int64, error) { var size int64 write := func(data []byte) error { if len(data) == 0 { @@ -290,16 +290,13 @@ func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw if err := checkUploadSize(size, maxSize); err != nil { return err } - if _, err := pw.Write(data); err != nil { - // A write only fails once the reader has gone away; surface it as the - // consumer-gone sentinel so streamUpload defers to the backend result - // rather than treating this as a client-side stream failure. - return errPipeConsumerGone + if _, err := w.Write(data); err != nil { + return fmt.Errorf("writing to staging file: %w", err) } return nil } - // Forward the data from the first request. + // Write the data from the first request. if err := write(firstData); err != nil { return size, err } @@ -320,14 +317,14 @@ func feedPipe(ctx context.Context, stream bytestream.ByteStream_WriteServer, pw return size, err } - // Forward this request's data first: a spec-compliant client may set + // Write this request's data first: a spec-compliant client may set // finish_write=true on the same message that carries the final chunk, // so the data must be written before the finish check or it is lost. if err := write(req.GetData()); err != nil { return size, err } - log.Debugw("msg", "upload chunk received (streaming)", "digest", digest, "currentSize", size, "maxSize", maxSize, "chunkSize", len(req.GetData())) + log.Debugw("msg", "upload chunk received", "digest", digest, "currentSize", size, "maxSize", maxSize, "chunkSize", len(req.GetData())) // Check if the client has finished sending data if req.GetFinishWrite() { @@ -395,85 +392,8 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream. return nil } -// Store the data received from the stream in a buffer and send a signal when finished -// This is done in a separate goroutine to avoid blocking the stream -func bufferStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, buffer *streamReader, log *log.Helper) { - // Send termination signal when finished receiving data - var bufferErr error - defer func() { - buffer.errorChan <- bufferErr - }() - - for { - select { - case <-ctx.Done(): - // DeadlineExceeded, or Canceled - bufferErr = ctx.Err() - return - default: - // Extract the next chunk of data from the stream request - req, err := getWriteRequest(stream) - if err != nil { - // If we have finished reading the stream we don't consider it a real error - if !errors.Is(err, io.EOF) { - bufferErr = err - } - return - } - - // Write the data first: a spec-compliant client may set - // finish_write=true on the same message that carries the final chunk, - // so the data must be buffered before the finish check or it is lost. - if err = buffer.Write(req.GetData()); err != nil { - bufferErr = err - return - } - - log.Debugw("msg", "upload chunk received", "digest", req.resource.Digest, "currentSize", buffer.size, "maxSize", buffer.maxSize, "chunkSize", len(req.GetData())) - - // Check if the client has finished sending data - if req.GetFinishWrite() { - return - } - } - } -} - -type streamReader struct { - *bytes.Buffer - // total size of the in-memory buffer in bytes - size int64 - // Max size allowed to be uploaded - maxSize int64 - // there was an error during stream data filling - errorChan chan error -} - -// Wrapper around a buffer that adds -// the ability to record the total size of the data that went through it -// and a channel to be used by the clients to signal when the buffer has been filled -func newStreamReader(maxSize int64) *streamReader { - return &streamReader{ - Buffer: bytes.NewBuffer(nil), - errorChan: make(chan error), - maxSize: maxSize, - } -} - -func (r *streamReader) Write(data []byte) error { - r.size += int64(len(data)) - - if err := checkUploadSize(r.size, r.maxSize); err != nil { - return err - } - - _, err := r.Buffer.Write(data) - return err -} - // checkUploadSize returns an ErrUploadSizeExceeded when total exceeds maxSize. -// maxSize == 0 means no limit. It is shared by the buffered (streamReader) and -// streaming (feedPipe) paths so their cap semantics cannot drift. +// maxSize == 0 means no limit. func checkUploadSize(total, maxSize int64) error { if maxSize != 0 && total > maxSize { return backend.NewErrUploadSizeExceeded(total, maxSize) diff --git a/app/artifact-cas/internal/service/bytestream_download_test.go b/app/artifact-cas/internal/service/bytestream_download_test.go index 95abb97ba..3025cab29 100644 --- a/app/artifact-cas/internal/service/bytestream_download_test.go +++ b/app/artifact-cas/internal/service/bytestream_download_test.go @@ -31,12 +31,10 @@ import ( "google.golang.org/grpc/codes" ) -// These tests lock down the DOWNLOAD digest-verification behavior. The download -// path is intentionally unchanged by the streaming-upload work (PFM-6923); this -// battery guards it against regressions — the server must stream the stored -// bytes back, compute their sha256 across however many chunks the backend -// produces, and reject any content whose digest does not match the requested -// resource name. +// These tests lock down the DOWNLOAD digest-verification behavior — the server +// must stream the stored bytes back, compute their sha256 across however many +// chunks the backend produces, and reject any content whose digest does not +// match the requested resource name. // fakeReadServer is a minimal bytestream.ByteStream_ReadServer that records the // data chunks the streamWriter sends. Only Send is exercised by streamWriter. diff --git a/app/artifact-cas/internal/service/bytestream_e2e_test.go b/app/artifact-cas/internal/service/bytestream_e2e_test.go new file mode 100644 index 000000000..6f05e4d0d --- /dev/null +++ b/app/artifact-cas/internal/service/bytestream_e2e_test.go @@ -0,0 +1,304 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net" + "os" + "path/filepath" + "runtime" + "strings" + "sync/atomic" + "testing" + "time" + + v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + jwtMiddleware "github.com/go-kratos/kratos/v2/middleware/auth/jwt" + + backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" + s3backend "github.com/chainloop-dev/chainloop/pkg/blobmanager/s3" + + grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth" + "github.com/minio/minio-go/v7" + miniocreds "github.com/minio/minio-go/v7/pkg/credentials" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" + + "github.com/go-kratos/kratos/v2/log" + "google.golang.org/genproto/googleapis/bytestream" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +// TestUploadEndToEndMinio drives the whole upload path against real +// infrastructure: a gRPC client talking over a TCP socket to the real +// ByteStreamService, which stages the artifact on a real filesystem and hands +// the resulting file to the real S3 backend writing into a real MinIO server. +// Nothing here is mocked except the credential provider, which only decides +// which backend to hand back. +// +// This is what the mock-backed suite cannot show: that the staged *os.File is +// something the AWS SDK actually accepts and uploads correctly, that the bytes +// landing in the object store are the bytes the client sent, and that a +// rejected upload leaves the bucket untouched. +func TestUploadEndToEndMinio(t *testing.T) { + if os.Getenv("SKIP_INTEGRATION") == "true" { + t.Skip() + } + + const bucket = "e2e-bucket" + endpoint := startMinio(t) + + minioClient, err := minio.New(endpoint, &minio.Options{ + Creds: miniocreds.NewStaticV4("root", "test-password", ""), Secure: false, + }) + require.NoError(t, err) + require.NoError(t, minioClient.MakeBucket(context.Background(), bucket, minio.MakeBucketOptions{})) + + realBackend, err := s3backend.NewBackend(&s3backend.Credentials{ + AccessKeyID: "root", + SecretAccessKey: "test-password", + Region: "us-east-1", + Location: fmt.Sprintf("http://%s/%s", endpoint, bucket), + }) + require.NoError(t, err) + + stagingDir := t.TempDir() + client, peakStagedBytes := newE2EClient(t, realBackend, stagingDir) + + t.Run("multipart upload lands byte-identical in the bucket", func(t *testing.T) { + // 12 MB forces the AWS SDK past its 5 MB part size, so the staged file + // is read back in parts rather than in one shot. + content := deterministicBytes(12 << 20) + digest := sha256Hex(content) + resource := &v1.CASResource{Digest: digest, FileName: "big-artifact.bin"} + + stream, err := client.Write(e2eUploadCtx()) + require.NoError(t, err) + sendInChunks(t, stream, encodeResource(t, resource), content, 64<<10) + + resp, err := stream.CloseAndRecv() + require.NoError(t, err) + require.Equal(t, int64(len(content)), resp.GetCommittedSize()) + + // The object really is in MinIO, and it really is the bytes we sent. + obj, err := minioClient.GetObject(context.Background(), bucket, "sha256:"+digest, minio.GetObjectOptions{}) + require.NoError(t, err) + t.Cleanup(func() { _ = obj.Close() }) + + stored := sha256.New() + written, err := copyInto(stored, obj) + require.NoError(t, err) + require.Equal(t, int64(len(content)), written, "stored object size must match what was uploaded") + require.Equal(t, digest, hex.EncodeToString(stored.Sum(nil)), + "the object stored under the canonical key must hash to that digest") + + require.Empty(t, listStaging(t, stagingDir), "staging dir must never hold an entry") + require.Positive(t, peakStagedBytes.Load(), + "the upload must actually spill to disk rather than buffer in memory") + require.Empty(t, openStagingFDs(t), "the staging descriptor must be released") + }) + + t.Run("digest mismatch is rejected and nothing is written to the bucket", func(t *testing.T) { + content := []byte("bytes that do not hash to the declared digest") + declared := sha256Hex([]byte("a completely different artifact")) + resource := &v1.CASResource{Digest: declared, FileName: "tampered.bin"} + + stream, err := client.Write(e2eUploadCtx()) + require.NoError(t, err) + sendInChunks(t, stream, encodeResource(t, resource), content, 8<<10) + + _, err = stream.CloseAndRecv() + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + require.Contains(t, err.Error(), "does not match the declared digest") + + // The canonical key must not exist: unverified bytes never reached S3. + _, err = minioClient.StatObject(context.Background(), bucket, "sha256:"+declared, minio.StatObjectOptions{}) + require.Error(t, err, "no object may be written under a digest that was never verified") + require.Equal(t, "NoSuchKey", minio.ToErrorResponse(err).Code) + + require.Empty(t, listStaging(t, stagingDir), "staging dir must never hold an entry") + require.Empty(t, openStagingFDs(t), "the staging descriptor must be released after a rejection") + }) +} + +// newE2EClient wires the real ByteStreamService to realBackend over a TCP gRPC +// connection. The returned counter records the largest staged artifact seen on +// disk while the server was running, so a test can prove the upload spilled +// rather than buffered. Staging files are unlinked at creation, so the size is +// read through the open descriptor rather than from the directory. +func newE2EClient(t *testing.T, realBackend backend.UploaderDownloader, stagingDir string) (bytestream.ByteStreamClient, *atomic.Int64) { + t.Helper() + + const backendType = "s3-e2e" + + provider := mocks.NewProvider(t) + provider.On("FromCredentials", mock.Anything, mock.Anything).Maybe().Return(realBackend, nil) + + server := grpc.NewServer( + grpc.StreamInterceptor( + grpc_auth.StreamServerInterceptor(func(ctx context.Context) (context.Context, error) { + return jwtMiddleware.NewContext(ctx, &casJWT.Claims{ + StoredSecretID: testStoredSecretID, + BackendType: backendType, + OrgID: testOrgID, + Role: casJWT.Uploader, + }), nil + }), + ), + ) + + bytestream.RegisterByteStreamServer(server, NewByteStreamService( + backend.Providers{backendType: provider}, + WithLogger(log.DefaultLogger), + WithAuditDispatcher(newTestDispatcher(&fakePublisher{})), + WithStagingDir(stagingDir), + )) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go func() { _ = server.Serve(lis) }() + t.Cleanup(server.Stop) + + conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + // Sample the staged artifact while the server runs so the test can assert it + // was really written to disk mid-upload. + var peak atomic.Int64 + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + go func() { + for { + select { + case <-stop: + return + default: + if n := stagedBytes(); n > peak.Load() { + peak.Store(n) + } + time.Sleep(time.Millisecond) + } + } + }() + + return bytestream.NewByteStreamClient(conn), &peak +} + +// listStaging returns the CAS staging files visible in the directory. Staging +// files are unlinked at creation, so this must stay empty at every moment of an +// upload, not only after one. +func listStaging(t *testing.T, dir string) []string { + t.Helper() + + entries, err := os.ReadDir(dir) + require.NoError(t, err) + + var found []string + for _, e := range entries { + if strings.HasPrefix(e.Name(), stagingFilePrefix) { + found = append(found, filepath.Join(dir, e.Name())) + } + } + return found +} + +// stagedBytes returns the size of the largest staging file this process holds +// open. The files are unlinked, so their size is only reachable through the +// descriptor; a non-zero reading during an upload is what proves the artifact +// went to disk instead of staying in memory. +func stagedBytes() int64 { + if runtime.GOOS != "linux" { + return 0 + } + + entries, err := os.ReadDir("/proc/self/fd") + if err != nil { + return 0 + } + + var largest int64 + for _, e := range entries { + fdPath := filepath.Join("/proc/self/fd", e.Name()) + target, err := os.Readlink(fdPath) + if err != nil || !strings.Contains(target, stagingFilePrefix) { + continue + } + // Stat through /proc/self/fd/N, which follows the descriptor rather than + // the (now absent) path. + info, err := os.Stat(fdPath) + if err != nil { + continue + } + if info.Size() > largest { + largest = info.Size() + } + } + return largest +} + +func e2eUploadCtx() context.Context { + return metadata.NewOutgoingContext(context.Background(), metadata.Pairs("role", "uploader")) +} + +func startMinio(t *testing.T) string { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + + const port = "9000/tcp" + instance, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: testcontainers.ContainerRequest{ + // Same pinned image the s3 backend suite uses. + Image: "quay.io/minio/minio@sha256:a1a8bd4ac40ad7881a245bab97323e18f971e4d4cba2c2007ec1bedd21cbaba2", + ExposedPorts: []string{port}, + Env: map[string]string{ + "MINIO_ROOT_USER": "root", + "MINIO_ROOT_PASSWORD": "test-password", + }, + Cmd: []string{"server", "/data"}, + WaitingFor: wait.ForListeningPort(port).WithStartupTimeout(5 * time.Minute), + }, + Started: true, + }) + require.NoError(t, err) + testcontainers.CleanupContainer(t, instance, testcontainers.StopTimeout(time.Minute)) + + p, err := instance.MappedPort(ctx, "9000") + require.NoError(t, err) + + return fmt.Sprintf("127.0.0.1:%d", p.Num()) +} + +// copyInto streams r into w, returning the number of bytes copied. +func copyInto(w io.Writer, r io.Reader) (int64, error) { + return io.Copy(w, r) +} diff --git a/app/artifact-cas/internal/service/bytestream_streaming_test.go b/app/artifact-cas/internal/service/bytestream_streaming_test.go index 777790e17..8770736fc 100644 --- a/app/artifact-cas/internal/service/bytestream_streaming_test.go +++ b/app/artifact-cas/internal/service/bytestream_streaming_test.go @@ -19,14 +19,18 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" + "os" + "path/filepath" + "runtime" + "strings" "syscall" "testing" "time" v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" - "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "google.golang.org/genproto/googleapis/bytestream" @@ -36,15 +40,6 @@ import ( const streamingBackendType = "streaming-backend-type" -// streamingUploaderDownloader wraps a mock backend and advertises streaming -// support, so the CAS service pipes the upload straight through instead of -// buffering it. Used only in tests to exercise the streaming code path. -type streamingUploaderDownloader struct { - *mocks.UploaderDownloader -} - -func (streamingUploaderDownloader) SupportsStreaming() bool { return true } - // --- test helpers --------------------------------------------------------- // streamingUpCtx returns an uploader context routed to the streaming backend. @@ -115,6 +110,190 @@ func (s *bytestreamSuite) expectStreamingUpload(resource *v1.CASResource, upload return received } +// --- upload integrity verification ----------------------------------------- + +// TestWriteDigestMismatchRejected is the core upload-integrity guarantee: bytes +// that do not hash to the client-declared digest are rejected with +// InvalidArgument, nothing is ever sent to the backend, and no audit event is +// emitted. Without verification an attacker could store arbitrary content under +// an arbitrary digest key. +func (s *bytestreamSuite) TestWriteDigestMismatchRejected() { + content := []byte("this is the real content that was actually streamed") + // The declared digest belongs to DIFFERENT content than what is sent. + resource := &v1.CASResource{Digest: sha256Hex([]byte("something else entirely")), FileName: "artifact.bin"} + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + // Upload must NOT be called for a mismatching digest. + + stream, err := s.client.Write(streamingUpCtx("")) + s.NoError(err) + sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 7) + + _, err = stream.CloseAndRecv() + assertGRPCError(s.T(), err, codes.InvalidArgument, "does not match the declared digest") + s.streamingBackend.AssertNotCalled(s.T(), "Upload", mock.Anything, mock.Anything, mock.Anything) + s.Empty(s.audit.published) +} + +// TestWriteBackendReceivesSeekableFile asserts the backend's Upload is handed a +// value satisfying io.ReaderAt+io.Seeker (an *os.File), so the AWS SDK's +// zero-buffer fast path is taken. Wrapping the file on the way to Upload (a +// TeeReader, progress reader, LimitReader) would silently reinstate part +// buffering, so this guards against that regression. +func (s *bytestreamSuite) TestWriteBackendReceivesSeekableFile() { + content := deterministicBytes(64 * 1024) + resource := resourceWithDigest(content, "artifact.bin") + var isReaderAt, isSeeker bool + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Run(func(args mock.Arguments) { + r := args.Get(1) + _, isReaderAt = r.(io.ReaderAt) + _, isSeeker = r.(io.Seeker) + _, _ = io.ReadAll(r.(io.Reader)) + }) + + stream, err := s.client.Write(streamingUpCtx("")) + s.NoError(err) + sendInChunks(s.T(), stream, encodeResource(s.T(), resource), content, 8192) + + _, err = stream.CloseAndRecv() + s.NoError(err) + s.True(isReaderAt, "backend must receive an io.ReaderAt for the SDK zero-buffer fast path") + s.True(isSeeker, "backend must receive an io.Seeker for the SDK zero-buffer fast path") +} + +// requireStagingEmpty asserts the per-test staging directory holds no files. +// Staging files are unlinked at creation, so this holds during an upload as well +// as after one; it is the directory-level half of the guarantee. +func (s *bytestreamSuite) requireStagingEmpty() { + entries, err := os.ReadDir(s.stagingDir) + s.Require().NoError(err) + s.Emptyf(entries, "staging dir must be left empty, found: %v", entries) +} + +// openStagingFDs reports the staging files this process still holds open. Since +// the files are unlinked at creation, the descriptor is the only reference left +// and closing it is what frees the space — so a leaked descriptor, not a leftover +// directory entry, is what would fill the staging volume. +// +// It reads /proc/self/fd, so it only reports on Linux; elsewhere it returns nil +// and the callers fall back to the directory-level assertion. +func openStagingFDs(t *testing.T) []string { + t.Helper() + + if runtime.GOOS != "linux" { + return nil + } + + entries, err := os.ReadDir("/proc/self/fd") + require.NoError(t, err) + + var held []string + for _, e := range entries { + target, err := os.Readlink(filepath.Join("/proc/self/fd", e.Name())) + if err != nil { + // The descriptor can vanish between listing and reading it. + continue + } + if strings.Contains(target, stagingFilePrefix) { + held = append(held, target) + } + } + return held +} + +// TestWriteWithoutStagingDirIsRefused: a service built without a staging +// directory must reject uploads outright. os.CreateTemp treats an empty path as +// the OS temp dir, which on the CAS container is the read-only /tmp secret mount, +// so the service has to refuse rather than stage somewhere unintended. +func (s *bytestreamSuite) TestWriteWithoutStagingDirIsRefused() { + svc := NewByteStreamService(nil, WithStagingDir("")) + _, err := svc.spillVerifyUpload(context.Background(), nil, nil, &writeRequest{ + resource: &v1.CASResource{Digest: sha256Hex([]byte("x")), FileName: "x.bin"}, + }, 0) + s.ErrorContains(err, "no staging directory configured") +} + +// TestWriteStagingCleanup: every exit path of the upload handler removes its +// staging file. Nothing else prunes the staging volume within the lifetime of a +// Pod — an emptyDir survives container restarts and is only cleared when the Pod +// is removed from the node — so a path that leaked here would accumulate until +// the volume filled. +func (s *bytestreamSuite) TestWriteStagingCleanup() { + content := []byte("staged, verified, then uploaded") + + testCases := []struct { + name string + // setup registers the backend expectations and returns the resource the + // client declares plus the bytes it streams. + setup func() (*v1.CASResource, []byte) + maxBytes string + expectCode codes.Code + expectMsg string + }{ + { + name: "upload succeeds", + setup: func() (*v1.CASResource, []byte) { + resource := resourceWithDigest(content, "ok.bin") + s.expectStreamingUpload(resource, nil) + return resource, content + }, + expectCode: codes.OK, + }, + { + name: "digest does not match the streamed bytes", + setup: func() (*v1.CASResource, []byte) { + resource := &v1.CASResource{Digest: sha256Hex([]byte("different")), FileName: "bad.bin"} + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + return resource, []byte("content that will not match the declared digest") + }, + expectCode: codes.InvalidArgument, + expectMsg: "does not match the declared digest", + }, + { + name: "backend upload fails", + setup: func() (*v1.CASResource, []byte) { + resource := resourceWithDigest(content, "backend-down.bin") + s.expectStreamingUpload(resource, errors.New("backend is down")) + return resource, content + }, + expectCode: codes.Internal, + }, + { + name: "upload exceeds the size cap", + setup: func() (*v1.CASResource, []byte) { + resource := resourceWithDigest(content, "too-big.bin") + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + return resource, content + }, + maxBytes: "4", + expectCode: codes.ResourceExhausted, + expectMsg: "max size of upload exceeded", + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + resource, sent := tc.setup() + + stream, err := s.client.Write(streamingUpCtx(tc.maxBytes)) + s.Require().NoError(err) + sendInChunks(s.T(), stream, encodeResource(s.T(), resource), sent, 8) + + _, err = stream.CloseAndRecv() + if tc.expectCode == codes.OK { + s.Require().NoError(err) + } else { + assertGRPCError(s.T(), err, tc.expectCode, tc.expectMsg) + } + + // The handler has returned by the time the client observes the final + // status, so its deferred close has already run. + s.requireStagingEmpty() + s.Emptyf(openStagingFDs(s.T()), "the staging descriptor must be closed, still open: %v", openStagingFDs(s.T())) + }) + } +} + // --- integrity / normal cases -------------------------------------------- // TestWriteStreamingSingleChunkOK: a single-chunk streaming upload stores the @@ -181,50 +360,6 @@ func (s *bytestreamSuite) TestWriteStreamingManyChunksIntegrity() { s.Len(gotBytes, len(content)) } -// TestWriteStreamingConsumesBeforeFinish is the core bounded-memory regression -// test (PFM-6923): the backend must begin consuming the upload BEFORE the client -// finishes sending. The handshake blocks the client's tail chunk until Upload -// has started; with a buffering implementation Upload is never entered until the -// whole stream is received, so this deadlocks and fails via timeout. -func (s *bytestreamSuite) TestWriteStreamingConsumesBeforeFinish() { - data := []byte("hello streaming world") - resource := resourceWithDigest(data, "artifact.bin") - - uploadStarted := make(chan struct{}) - received := make(chan []byte, 1) - s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource). - Return(nil).Run(func(args mock.Arguments) { - close(uploadStarted) - got, err := io.ReadAll(args.Get(1).(io.Reader)) - s.NoError(err) - received <- got - }) - - stream, err := s.client.Write(streamingUpCtx("")) - s.NoError(err) - s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), resource), - Data: data[:6], - })) - - select { - case <-uploadStarted: - case <-time.After(5 * time.Second): - s.FailNow("backend Upload was not started before the stream finished — upload is being buffered, not streamed") - } - - s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), resource), - Data: data[6:], - })) - - got, err := stream.CloseAndRecv() - s.NoError(err) - s.Equal(int64(len(data)), got.CommittedSize) - s.Equal(data, <-received) -} - // TestWriteStreamingEmptyArtifact: a zero-byte artifact streams cleanly and is // committed with size 0. func (s *bytestreamSuite) TestWriteStreamingEmptyArtifact() { @@ -359,8 +494,10 @@ func (s *bytestreamSuite) TestWriteStreamingMaxSizeUnlimited() { // TestWriteStreamingBackendError: a generic backend Upload failure surfaces as // Internal and emits no audit event. func (s *bytestreamSuite) TestWriteStreamingBackendError() { - s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource). + data := []byte("hello world") + resource := resourceWithDigest(data, "artifact.bin") + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource). Return(fmt.Errorf("object store rejected the upload")).Run(func(args mock.Arguments) { _, _ = io.ReadAll(args.Get(1).(io.Reader)) }) @@ -368,8 +505,8 @@ func (s *bytestreamSuite) TestWriteStreamingBackendError() { stream, err := s.client.Write(streamingUpCtx("")) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -381,16 +518,18 @@ func (s *bytestreamSuite) TestWriteStreamingBackendError() { // wrapping a network reset must be masked as Internal, NOT mistaken for a client // disconnect (which would falsely report success and silently drop the blob). func (s *bytestreamSuite) TestWriteStreamingBackendResetNotTreatedAsDisconnect() { - s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource). + data := []byte("hello world") + resource := resourceWithDigest(data, "artifact.bin") + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource). Return(fmt.Errorf("connection to object store failed: %w", syscall.ECONNRESET)). Run(func(args mock.Arguments) { _, _ = io.ReadAll(args.Get(1).(io.Reader)) }) stream, err := s.client.Write(streamingUpCtx("")) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -403,16 +542,18 @@ func (s *bytestreamSuite) TestWriteStreamingBackendResetNotTreatedAsDisconnect() // TestWriteStreamingBackendCanceledNotTreatedAsDisconnect: same guard for an // Upload error wrapping context.Canceled originating backend-side. func (s *bytestreamSuite) TestWriteStreamingBackendCanceledNotTreatedAsDisconnect() { - s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource). + data := []byte("hello world") + resource := resourceWithDigest(data, "artifact.bin") + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource). Return(fmt.Errorf("backend deadline: %w", context.Canceled)). Run(func(args mock.Arguments) { _, _ = io.ReadAll(args.Get(1).(io.Reader)) }) stream, err := s.client.Write(streamingUpCtx("")) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -425,13 +566,14 @@ func (s *bytestreamSuite) TestWriteStreamingBackendCanceledNotTreatedAsDisconnec // service's own reader-close must not surface as an Internal error. func (s *bytestreamSuite) TestWriteStreamingBackendSuccessWithoutDrain() { data := []byte("hello world") - s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(nil) // no drain + resource := resourceWithDigest(data, "artifact.bin") + s.streamingBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.streamingBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil) // no drain stream, err := s.client.Write(streamingUpCtx("")) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), + ResourceName: encodeResource(s.T(), resource), Data: data, })) @@ -445,18 +587,12 @@ func (s *bytestreamSuite) TestWriteStreamingBackendSuccessWithoutDrain() { } // TestWriteStreamingClientDisconnect: when the client cancels mid-upload, the -// server treats it as a disconnect (no error masking, no audit) and the backend -// sees the stream abort through the pipe. +// server aborts before verification completes, so nothing is ever sent to the +// backend and no audit event is emitted. func (s *bytestreamSuite) TestWriteStreamingClientDisconnect() { - uploadStarted := make(chan struct{}) - readErr := make(chan error, 1) - s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.streamingBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Maybe(). - Return(nil).Run(func(args mock.Arguments) { - close(uploadStarted) - _, err := io.ReadAll(args.Get(1).(io.Reader)) - readErr <- err - }) + // Exists may or may not be reached depending on how fast the cancellation + // races the handler; the invariant under test is that Upload never is. + s.streamingBackend.On("Exists", mock.Anything, s.resource.Digest).Maybe().Return(false, nil) ctx, cancel := context.WithCancel(streamingUpCtx("")) stream, err := s.client.Write(ctx) @@ -465,26 +601,22 @@ func (s *bytestreamSuite) TestWriteStreamingClientDisconnect() { ResourceName: encodeResource(s.T(), s.resource), Data: []byte("partial upload"), })) - - // Wait until the backend is actively consuming, then cancel the client so the - // disconnect happens deterministically mid-stream. - select { - case <-uploadStarted: - case <-time.After(5 * time.Second): - s.FailNow("backend Upload was not started") - } cancel() - select { - case err := <-readErr: - // The backend saw the aborted stream (pipe closed with the cancellation). - s.Error(err) - case <-time.After(5 * time.Second): - s.FailNow("backend Upload did not observe the client disconnect") - } - - // A disconnect is not a successful upload: no audit event is emitted. + _, err = stream.CloseAndRecv() + // The client canceled: the RPC ends in error and nothing is stored. + s.Error(err) s.Empty(s.audit.published) + s.streamingBackend.AssertNotCalled(s.T(), "Upload", mock.Anything, mock.Anything, mock.Anything) + + // The staging file is removed here too. Cancelling surfaces the error to the + // client as soon as it happens, which can race the handler's own unwinding, + // so wait for the handler to finish rather than sampling the directory right + // away. + s.Require().Eventually(func() bool { + entries, err := os.ReadDir(s.stagingDir) + return err == nil && len(entries) == 0 && len(openStagingFDs(s.T())) == 0 + }, 5*time.Second, 10*time.Millisecond, "the staging descriptor must be released after a client disconnect") } // --- dedup ---------------------------------------------------------------- diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go index 096a467ea..8fc6d0adb 100644 --- a/app/artifact-cas/internal/service/bytestream_test.go +++ b/app/artifact-cas/internal/service/bytestream_test.go @@ -48,42 +48,6 @@ import ( "google.golang.org/grpc/test/bufconn" ) -func (s *bytestreamSuite) TestStreamReader() { - buffer := newStreamReader(0) - // Write twice and check the length - err := buffer.Write([]byte("hello")) - s.NoError(err) - s.Equal(int64(5), buffer.size) - err = buffer.Write([]byte("chainloop")) - s.NoError(err) - s.Equal(int64(14), buffer.size) - // The buffer length also matches - s.Equal(14, buffer.Len()) - - // Start reading - writer := bytes.NewBuffer(nil) - copied, err := io.Copy(writer, buffer) - s.Equal(int64(14), copied) - s.NoError(err) - // The buffer length is still 14 to indicate what it has processed - s.Equal(int64(14), buffer.size) - // but the internal one is 0 - s.Equal(0, buffer.Len()) -} - -func (s *bytestreamSuite) TestStreamReaderOverflow() { - // a buffer with 8 bytes limit - buffer := newStreamReader(8) - // Write twice and check the length - err := buffer.Write([]byte("hello")) - s.NoError(err) - s.Equal(int64(5), buffer.size) - err = buffer.Write([]byte("chainloop")) - s.Error(err) - s.True(backend.IsUploadSizeExceeded(err)) - s.ErrorContains(err, "max size of upload exceeded") -} - func (s *bytestreamSuite) TestWrite() { ctx := s.upCtx @@ -210,19 +174,20 @@ func (s *bytestreamSuite) TestWriteExistInternalTraffic() { func (s *bytestreamSuite) TestWriteOK() { data := []byte("hello world") - s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(nil) + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil) stream, err := s.client.Write(s.upCtx) s.NoError(err) // Multiple chunks s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), + ResourceName: encodeResource(s.T(), resource), Data: data[:5], })) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), + ResourceName: encodeResource(s.T(), resource), Data: data[5:], })) @@ -234,20 +199,22 @@ func (s *bytestreamSuite) TestWriteOK() { s.Require().Len(s.audit.published, 1) info := decodeArtifactEvent(s.T(), s.audit.published[0]) s.False(info.Skipped) - s.Equal(s.resource.Digest, info.Digest) + s.Equal(resource.Digest, info.Digest) s.Equal(int64(len(data)), info.SizeBytes) - s.Equal(s.resource.FileName, info.FileName) + s.Equal(resource.FileName, info.FileName) } func (s *bytestreamSuite) TestWriteErrorUploading() { - s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource).Return(errors.New("error uploading")) + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(errors.New("error uploading")) stream, err := s.client.Write(s.upCtx) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -256,9 +223,9 @@ func (s *bytestreamSuite) TestWriteErrorUploading() { s.Empty(s.audit.published) } -// TestWriteBufferedMultiChunkIntegrity: the buffered/OCI path reassembles a -// multi-chunk upload byte-for-byte and hands the backend content whose sha256 -// matches the declared digest — parity with the streaming path. +// TestWriteBufferedMultiChunkIntegrity: an OCI-backed multi-chunk upload is +// reassembled byte-for-byte and the backend receives content whose sha256 +// matches the declared digest. func (s *bytestreamSuite) TestWriteBufferedMultiChunkIntegrity() { content := []byte("chainloop attestation payload spanning multiple stream chunks") resource := resourceWithDigest(content, "artifact.bin") @@ -314,15 +281,17 @@ func (s *bytestreamSuite) TestWriteBufferedFinishWriteWithData() { // masks a backend-side failure wrapping a network reset as Internal, never // mistaking it for a client disconnect (which would falsely report success). func (s *bytestreamSuite) TestWriteBufferedBackendResetNotTreatedAsDisconnect() { - s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource). + data := []byte("hello world") + resource := resourceWithDigest(data, "artifact.bin") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource). Return(fmt.Errorf("connection to registry failed: %w", syscall.ECONNRESET)) stream, err := s.client.Write(s.upCtx) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -335,15 +304,17 @@ func (s *bytestreamSuite) TestWriteBufferedBackendResetNotTreatedAsDisconnect() // TestWriteBufferedBackendCanceledNotTreatedAsDisconnect: same guard for an // Upload error wrapping context.Canceled originating backend-side. func (s *bytestreamSuite) TestWriteBufferedBackendCanceledNotTreatedAsDisconnect() { - s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(false, nil) - s.ociBackend.On("Upload", mock.Anything, mock.Anything, s.resource). + data := []byte("hello world") + resource := resourceWithDigest(data, "artifact.bin") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource). Return(fmt.Errorf("registry deadline: %w", context.Canceled)) stream, err := s.client.Write(s.upCtx) s.NoError(err) s.NoError(stream.Send(&bytestream.WriteRequest{ - ResourceName: encodeResource(s.T(), s.resource), - Data: []byte("hello world"), + ResourceName: encodeResource(s.T(), resource), + Data: data, })) _, err = stream.CloseAndRecv() @@ -457,13 +428,16 @@ type bytestreamSuite struct { srv *grpc.Server client bytestream.ByteStreamClient ociBackend *mocks.UploaderDownloader - // streamingBackend is a streaming-capable backend (implements - // backend.StreamingUploader) reachable via the "backend-streaming" metadata. + // streamingBackend is the backend reachable via the "backend-streaming" + // metadata. streamingBackend *mocks.UploaderDownloader resource *v1.CASResource audit *fakePublisher upCtx context.Context downCtx context.Context + // stagingDir is the per-test upload staging directory the service is + // configured with, so tests can assert it is left clean. + stagingDir string } // Run after each test @@ -472,6 +446,16 @@ func (s *bytestreamSuite) TearDownTest() { s.srv.Stop() } +// SetupSubTest and TearDownSubTest give every s.Run subtest its own server, +// backend mocks and staging directory, so table-driven cases stay isolated. +func (s *bytestreamSuite) SetupSubTest() { + s.SetupTest() +} + +func (s *bytestreamSuite) TearDownSubTest() { + s.TearDownTest() +} + func (s *bytestreamSuite) SetupTest() { const backendType = "backend-type" // 1 MB buffer @@ -522,21 +506,21 @@ func (s *bytestreamSuite) SetupTest() { ociBackend := mocks.NewUploaderDownloader(s.T()) ociBackendProvider.On("FromCredentials", mock.Anything, mock.Anything).Maybe().Return(ociBackend, nil) - // A streaming-capable backend (object stores like S3/Azure). It wraps a mock - // so tests can set expectations on it while the service detects it as - // streaming via the backend.StreamingUploader interface. + // A second, object-store-like backend reachable via the "backend-streaming" + // metadata that tests can set expectations on. streamingBackend := mocks.NewUploaderDownloader(s.T()) streamingBackendProvider := mocks.NewProvider(s.T()) streamingBackendProvider.On("FromCredentials", mock.Anything, mock.Anything).Maybe(). - Return(&streamingUploaderDownloader{streamingBackend}, nil) + Return(streamingBackend, nil) s.audit = &fakePublisher{} + s.stagingDir = s.T().TempDir() bytestream.RegisterByteStreamServer( server, NewByteStreamService(backend.Providers{ backendType: ociBackendProvider, streamingBackendType: streamingBackendProvider, - }, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit))), + }, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)), ) go func() { _ = server.Serve(l) diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index a1a24c2ef..3f055c3c9 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -38,6 +38,11 @@ type commonService struct { backends backend.Providers // best-effort audit events publisher, nil-safe audit *AuditDispatcher + // stagingDir is the local directory where uploads are staged on disk while + // their SHA256 is verified against the declared digest before reaching the + // backend. It must be set and writable: leaving it empty is a deployment + // error and uploads are refused rather than staged somewhere unintended. + stagingDir string } func (s *commonService) loadBackend(ctx context.Context, providerType, secretID string) (backend.UploaderDownloader, error) { @@ -72,6 +77,16 @@ func WithAuditDispatcher(d *AuditDispatcher) NewOpt { } } +// WithStagingDir sets the local directory where uploads are spilled and +// verified before being sent to the backend. It must point at a writable volume +// dedicated to this pod; there is no default, so an upload fails loudly rather +// than silently staging somewhere unintended. +func WithStagingDir(dir string) NewOpt { + return func(s *commonService) { + s.stagingDir = dir + } +} + func newCommonService(backends backend.Providers, opts ...NewOpt) *commonService { s := &commonService{ log: servicelogger.EmptyLogger(), diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index d85a23845..eb02b30bc 100644 --- a/deployment/chainloop/Chart.yaml +++ b/deployment/chainloop/Chart.yaml @@ -7,7 +7,7 @@ description: Chainloop is an open source software supply chain control plane, a type: application # Bump the patch (not minor, not major) version on each change in the Chart Source code -version: 1.441.0 +version: 1.441.1 # Do not update appVersion, this is handled automatically by the release process appVersion: v1.109.4 diff --git a/deployment/chainloop/templates/_helpers.tpl b/deployment/chainloop/templates/_helpers.tpl index 7dcd8e264..4665545e0 100644 --- a/deployment/chainloop/templates/_helpers.tpl +++ b/deployment/chainloop/templates/_helpers.tpl @@ -492,3 +492,13 @@ Return the Nats connection string for the CAS {{- $port := required "nats server port not set" .Values.cas.nats.port }} {{- printf "nats://%s:%d" $host ($port | int) }} {{- end -}} + +{{/* +Directory where the CAS stages and verifies uploads before sending them to the +backend. Not configurable: it must be the dedicated per-pod emptyDir mounted by +the CAS deployment, and it must not collide with /tmp, which the jwt-public-key +secret already mounts read-only. +*/}} +{{- define "chainloop.cas.staging_dir" -}} +/tmp-staging-fs +{{- end -}} diff --git a/deployment/chainloop/templates/cas/configmap.yaml b/deployment/chainloop/templates/cas/configmap.yaml index ae17ba695..402bef8fb 100644 --- a/deployment/chainloop/templates/cas/configmap.yaml +++ b/deployment/chainloop/templates/cas/configmap.yaml @@ -14,6 +14,9 @@ metadata: {{- end }} data: server.yaml: | + # Local directory where uploads are staged and verified against the declared + # digest before being sent to the backend. Backed by the staging emptyDir. + staging_dir: {{ include "chainloop.cas.staging_dir" . | quote }} server: http: addr: "0.0.0.0:{{ .Values.cas.containerPorts.http }}" diff --git a/deployment/chainloop/templates/cas/deployment.yaml b/deployment/chainloop/templates/cas/deployment.yaml index 96a57e2de..b99892cc2 100644 --- a/deployment/chainloop/templates/cas/deployment.yaml +++ b/deployment/chainloop/templates/cas/deployment.yaml @@ -126,6 +126,10 @@ spec: mountPath: "/data/conf" - name: jwt-public-key mountPath: "/tmp" + # Writable scratch volume for staging + verifying uploads before they + # reach the backend (the container root filesystem is read-only). + - name: staging + mountPath: {{ include "chainloop.cas.staging_dir" . | quote }} {{- if eq "gcpSecretManager" .Values.secretsBackend.backend }} - name: gcp-secretmanager-serviceaccountkey mountPath: /gcp-secrets @@ -158,6 +162,11 @@ spec: - name: jwt-public-key secret: secretName: {{ include "chainloop.cas.fullname" . }}-jwt-public-key + # Node-disk (NOT tmpfs/RAM) scratch space for staging + verifying uploads. + # No sizeLimit: breaching one evicts the Pod rather than failing the + # upload that overran it. + - name: staging + emptyDir: {} {{- if include "cas.tls-secret-name" . }} - name: server-certs secret: diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index a3453df85..50c604c66 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.yaml @@ -1478,7 +1478,21 @@ cas: drop: ["ALL"] seccompProfile: type: "RuntimeDefault" - + + ## CAS upload staging. + ## Uploads are streamed to a directory on local disk and verified against the + ## declared digest before being sent to the storage backend; nothing unverified + ## ever reaches the backend. The CAS mounts a dedicated emptyDir for this and + ## points staging_dir at it. There is nothing to configure: the path is fixed + ## because it must not collide with /tmp, which the jwt-public-key secret + ## already mounts read-only, and the volume must stay per-pod. + ## + ## The volume is sized by the node's ephemeral storage rather than by a + ## sizeLimit on the emptyDir: exceeding a sizeLimit gets the Pod EVICTED by the + ## kubelet instead of failing the offending upload. Bound uploads with the + ## per-request size cap, and set cas.resources ephemeral-storage requests to + ## have the scheduler account for the space. + ## @param cas.automountServiceAccountToken Mount Service Account token in cas pods ## automountServiceAccountToken: false diff --git a/pkg/blobmanager/azureblob/backend.go b/pkg/blobmanager/azureblob/backend.go index d70893553..edc3f72b4 100644 --- a/pkg/blobmanager/azureblob/backend.go +++ b/pkg/blobmanager/azureblob/backend.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "io" + "os" "github.com/Azure/azure-sdk-for-go/sdk/azcore/to" "github.com/Azure/azure-sdk-for-go/sdk/azidentity" @@ -37,15 +38,7 @@ type Backend struct { endpoint string } -var ( - _ backend.UploaderDownloader = (*Backend)(nil) - _ backend.StreamingUploader = (*Backend)(nil) -) - -// SupportsStreaming reports that the azureblob backend can upload directly from -// a streaming reader. Upload uses the SDK's UploadStream, which reads the artifact -// in bounded-size blocks, so CAS never needs to buffer the whole blob in memory. -func (b *Backend) SupportsStreaming() bool { return true } +var _ backend.UploaderDownloader = (*Backend)(nil) func NewBackend(creds *Credentials) (*Backend, error) { credential, err := azidentity.NewClientSecretCredential(creds.TenantID, creds.ClientID, creds.ClientSecret, nil) @@ -128,11 +121,25 @@ func (b *Backend) Upload(ctx context.Context, r io.Reader, resource *pb.CASResou return fmt.Errorf("failed to create Blob storage Container: %w", err) } + metadata := map[string]*string{ + annotationNameAuthor: to.Ptr(backend.AuthorAnnotation), + annotationNameFilename: to.Ptr(resource.FileName), + } + + // The CAS service stages verified content on local disk and hands us an + // *os.File. UploadFile reads it via ReadAt in bounded blocks, avoiding the + // intermediate block buffering that UploadStream needs for a non-seekable + // reader. Fall back to UploadStream for any other reader (still bounded by + // block size × concurrency). + if f, ok := r.(*os.File); ok { + _, err = client.UploadFile(ctx, b.container, resourceName(resource.Digest), f, &azblob.UploadFileOptions{ + Metadata: metadata, + }) + return err + } + _, err = client.UploadStream(ctx, b.container, resourceName(resource.Digest), r, &azblob.UploadStreamOptions{ - Metadata: map[string]*string{ - annotationNameAuthor: to.Ptr(backend.AuthorAnnotation), - annotationNameFilename: to.Ptr(resource.FileName), - }, + Metadata: metadata, }) return err diff --git a/pkg/blobmanager/azureblob/backend_test.go b/pkg/blobmanager/azureblob/backend_test.go deleted file mode 100644 index f3a17b20d..000000000 --- a/pkg/blobmanager/azureblob/backend_test.go +++ /dev/null @@ -1,34 +0,0 @@ -// -// Copyright 2026 The Chainloop Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package azureblob - -import ( - "testing" - - backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// TestBackend_SupportsStreaming asserts the azureblob backend opts into -// streaming uploads so the CAS service feeds it directly from the client stream -// instead of buffering the whole artifact in memory (PFM-6923). -func TestBackend_SupportsStreaming(t *testing.T) { - var b backend.UploaderDownloader = &Backend{} - su, ok := b.(backend.StreamingUploader) - require.True(t, ok, "azureblob backend must implement backend.StreamingUploader") - assert.True(t, su.SupportsStreaming()) -} diff --git a/pkg/blobmanager/backend.go b/pkg/blobmanager/backend.go index c4c4f355a..03c868b04 100644 --- a/pkg/blobmanager/backend.go +++ b/pkg/blobmanager/backend.go @@ -44,22 +44,6 @@ type UploaderDownloader interface { Describer } -// StreamingUploader is an optional interface implemented by backends whose -// Upload can consume the artifact directly from a streaming io.Reader without -// requiring the whole blob to be buffered in memory first. -// -// The Artifact CAS service type-asserts uploaders against this interface: when -// a backend reports SupportsStreaming()==true the upload is piped straight from -// the client stream to the backend, bounding CAS memory usage independently of -// artifact size (PFM-6923). Backends that do not implement it (e.g. the OCI -// backend, whose push path needs the full layer content up front) keep the -// buffered code path. -type StreamingUploader interface { - // SupportsStreaming reports whether Upload can be fed a streaming reader - // without the caller buffering the full artifact in memory first. - SupportsStreaming() bool -} - type Describer interface { Describe(ctx context.Context, digest string) (*v1.CASResource, error) } diff --git a/pkg/blobmanager/oci/backend_test.go b/pkg/blobmanager/oci/backend_test.go index c0d47638d..4b509b30d 100644 --- a/pkg/blobmanager/oci/backend_test.go +++ b/pkg/blobmanager/oci/backend_test.go @@ -27,7 +27,6 @@ import ( "testing" pb "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" - backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" "github.com/google/go-containerregistry/pkg/authn" "github.com/google/go-containerregistry/pkg/name" "github.com/google/go-containerregistry/pkg/registry" @@ -409,14 +408,3 @@ func (s *testSuite) TearDownTest() { func TestOCIBackend(t *testing.T) { suite.Run(t, new(testSuite)) } - -// TestBackend_DoesNotSupportStreaming pins the OCI backend as NON-streaming. -// go-containerregistry's push path needs the whole layer content in memory up -// front (see Backend.Upload), so the CAS service must keep buffering OCI -// uploads. If OCI ever grows a SupportsStreaming method it would be fed a -// streaming reader and silently break; this test fails closed against that. -func TestBackend_DoesNotSupportStreaming(t *testing.T) { - var b backend.UploaderDownloader = &Backend{} - _, ok := b.(backend.StreamingUploader) - require.False(t, ok, "oci backend must NOT implement backend.StreamingUploader; it requires full in-memory buffering") -} diff --git a/pkg/blobmanager/s3/backend.go b/pkg/blobmanager/s3/backend.go index 8fdc9546a..f9b8c8f3e 100644 --- a/pkg/blobmanager/s3/backend.go +++ b/pkg/blobmanager/s3/backend.go @@ -46,19 +46,10 @@ type Backend struct { customEndpoint string } -var ( - _ backend.UploaderDownloader = (*Backend)(nil) - _ backend.StreamingUploader = (*Backend)(nil) -) +var _ backend.UploaderDownloader = (*Backend)(nil) const defaultRegion = "us-east-1" -// SupportsStreaming reports that the s3 backend can upload directly from a -// streaming reader. The AWS SDK's manager.Uploader consumes the reader in -// bounded-size parts (multipart upload), so CAS never needs to buffer the whole -// artifact in memory. -func (b *Backend) SupportsStreaming() bool { return true } - func NewBackend(creds *Credentials) (*Backend, error) { if creds == nil { return nil, errors.New("credentials cannot be nil") @@ -173,10 +164,9 @@ func (b *Backend) Upload(ctx context.Context, r io.Reader, resource *pb.CASResou }, } - // if b.checksumVerificationEnabled() { - // // Check that the object is uploaded correctly - // input.ChecksumSHA256 = aws.String(hexSha256ToBinaryB64(resource.Digest)) - // } + // No ChecksumSHA256 precondition: a whole-object SHA256 cannot be expressed + // for a multipart upload (S3 offers FULL_OBJECT checksums only for the CRC + // variants), and some S3-compatible endpoints such as R2 reject it outright. if _, err := uploader.Upload(ctx, input); err != nil { return fmt.Errorf("failed to upload to bucket: %w", err) diff --git a/pkg/blobmanager/s3/backend_test.go b/pkg/blobmanager/s3/backend_test.go index e936378e2..a61495e82 100644 --- a/pkg/blobmanager/s3/backend_test.go +++ b/pkg/blobmanager/s3/backend_test.go @@ -244,12 +244,6 @@ func (s *testSuite) TestDownload() { s.NoError(err) s.Equal("test", buf.String()) }) - - // s.T().Run("it's been tampered", func(t *testing.T) { - // buf := bytes.NewBuffer(nil) - // err := s.backend.Download(context.Background(), buf, s.tamperedObjectDigest) - // s.ErrorContains(err, "failed to validate integrity of object") - // }) } type testSuite struct { @@ -258,7 +252,6 @@ type testSuite struct { backend, invalidBackend *Backend ownedObjectDigest string externalObjectDigest string - tamperedObjectDigest string } func TestS3Backend(t *testing.T) { @@ -313,15 +306,6 @@ func (s *testSuite) SetupTest() { err = s.backend.Upload(context.Background(), buf, &pb.CASResource{Digest: s.ownedObjectDigest, FileName: "test.txt"}) require.NoError(s.T(), err) - // Copy an existing object but reference it from somewhere else - s.tamperedObjectDigest = "b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c" - _, err = minioClient.CopyObject(context.Background(), minio.CopyDestOptions{ - Bucket: testBucket, Object: fmt.Sprintf("sha256:%s", s.tamperedObjectDigest), - }, minio.CopySrcOptions{ - Bucket: testBucket, Object: fmt.Sprintf("sha256:%s", s.ownedObjectDigest), - }) - require.NoError(s.T(), err) - // upload another one but by the client directly reader := bytes.NewReader([]byte("hello world")) s.externalObjectDigest = "external-deadbeef" @@ -378,13 +362,3 @@ func (c *minioInstance) ConnectionString(t *testing.T) string { type minioInstance struct { instance testcontainers.Container } - -// TestBackend_SupportsStreaming asserts the s3 backend opts into streaming -// uploads so the CAS service feeds it directly from the client stream instead -// of buffering the whole artifact in memory (PFM-6923). -func TestBackend_SupportsStreaming(t *testing.T) { - var b backend.UploaderDownloader = &Backend{} - su, ok := b.(backend.StreamingUploader) - require.True(t, ok, "s3 backend must implement backend.StreamingUploader") - assert.True(t, su.SupportsStreaming()) -} diff --git a/pkg/blobmanager/s3accesspoint/backend.go b/pkg/blobmanager/s3accesspoint/backend.go index 71d2524cc..6c30e663a 100644 --- a/pkg/blobmanager/s3accesspoint/backend.go +++ b/pkg/blobmanager/s3accesspoint/backend.go @@ -73,16 +73,7 @@ type Backend struct { s3Client *s3.Client } -var ( - _ backend.UploaderDownloader = (*Backend)(nil) - _ backend.StreamingUploader = (*Backend)(nil) -) - -// SupportsStreaming reports that the s3accesspoint backend can upload directly -// from a streaming reader. Like the plain s3 backend it uses the AWS SDK's -// manager.Uploader, which consumes the reader in bounded-size multipart parts, -// so CAS never needs to buffer the whole artifact in memory. -func (b *Backend) SupportsStreaming() bool { return true } +var _ backend.UploaderDownloader = (*Backend)(nil) // NewBackend constructs a *Backend wired to an STS-backed credentials // provider. ctx is used only for the initial AWS config load (DNS lookups, diff --git a/pkg/blobmanager/s3accesspoint/backend_test.go b/pkg/blobmanager/s3accesspoint/backend_test.go index 1b929bd7e..57cbd34a8 100644 --- a/pkg/blobmanager/s3accesspoint/backend_test.go +++ b/pkg/blobmanager/s3accesspoint/backend_test.go @@ -26,7 +26,6 @@ import ( ststypes "github.com/aws/aws-sdk-go-v2/service/sts/types" pb "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" robotaccount "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" - backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" jwtmiddleware "github.com/go-kratos/kratos/v2/middleware/auth/jwt" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -295,13 +294,3 @@ func assertFailedClosed(t *testing.T, err error) { require.Containsf(t, err.Error(), ErrMissingRequestingOrg.Error(), "expected fail-closed missing-org error, got %q", err) } - -// TestBackend_SupportsStreaming asserts the s3accesspoint backend opts into -// streaming uploads so the CAS service feeds it directly from the client stream -// instead of buffering the whole artifact in memory (PFM-6923). -func TestBackend_SupportsStreaming(t *testing.T) { - var b backend.UploaderDownloader = &Backend{} - su, ok := b.(backend.StreamingUploader) - require.True(t, ok, "s3accesspoint backend must implement backend.StreamingUploader") - assert.True(t, su.SupportsStreaming()) -}