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()) -}