diff --git a/app/artifact-cas/cmd/main.go b/app/artifact-cas/cmd/main.go index 5e64b440d..1e7aa4a69 100644 --- a/app/artifact-cas/cmd/main.go +++ b/app/artifact-cas/cmd/main.go @@ -123,8 +123,8 @@ 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. + // Ensure the staging directory is configured and writable before we accept + // any traffic: without it no upload or download can be verified. if err := prepareStagingDir(&bc); err != nil { panic(err) } @@ -160,20 +160,20 @@ 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). +// prepareStagingDir creates the 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. +// staging_dir is required: uploads and downloads are verified by spilling them +// here first, so a missing or unwritable directory means no transfer can +// succeed. Failing at startup surfaces that immediately, rather than letting the +// service report healthy and reject every request. 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 artifacts 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. +// The directory stays clean on its own: each request unlinks its staging file as +// soon as it is created, 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 == "" { diff --git a/app/artifact-cas/configs/config.devel.yaml b/app/artifact-cas/configs/config.devel.yaml index 50c84adb6..65936c30c 100644 --- a/app/artifact-cas/configs/config.devel.yaml +++ b/app/artifact-cas/configs/config.devel.yaml @@ -40,7 +40,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. +# Local directory where uploads and downloads are staged and verified against +# their digest before being forwarded (to the backend, or to the client). +# 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 c364830d7..ac72ca7ae 100644 --- a/app/artifact-cas/internal/conf/conf.pb.go +++ b/app/artifact-cas/internal/conf/conf.pb.go @@ -47,11 +47,11 @@ type Bootstrap struct { // 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"` - // 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. + // Local directory where uploads and downloads are staged on disk and verified + // against their digest: uploads before reaching the backend, downloads before + // the first byte is sent to the client. It must be a writable volume; in + // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the + // /tmp secret mount). Required: when unset the service refuses to start. StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto index 5c14b4227..d9c8c5399 100644 --- a/app/artifact-cas/internal/conf/conf.proto +++ b/app/artifact-cas/internal/conf/conf.proto @@ -28,11 +28,11 @@ 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. + // Local directory where uploads and downloads are staged on disk and verified + // against their digest: uploads before reaching the backend, downloads before + // the first byte is sent to the client. It must be a writable volume; in + // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the + // /tmp secret mount). Required: when unset the service refuses to start. string staging_dir = 6; message NatsServer { diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 8ea5f2b51..6a4cbe5f7 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -18,16 +18,11 @@ package service import ( "bytes" "context" - "crypto/sha256" "encoding/base64" "encoding/gob" - "encoding/hex" + "errors" "fmt" - "hash" "io" - "os" - - "errors" v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor/events" @@ -59,8 +54,9 @@ func NewByteStreamService(bp backend.Providers, opts ...NewOpt) *ByteStreamServi } // Client-side streaming RPC for writing blobs. -// Iterate on the stream of file chunks, aggregate them in a buffer, -// send them to the backend and return a response with the commitedSize +// Iterate on the stream of file chunks, spill them to the staging disk, verify +// them against the declared digest, hand the verified file to the backend and +// return a response with the committedSize func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) error { ctx := stream.Context() ctx, span := otelx.Start(ctx, byteStreamTracer, "ByteStreamService.Write") @@ -177,68 +173,14 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro // 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) - } - - // 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()) - } - }() - - // 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) + f, size, err := s.stageAndVerify(stagingUpload, req.resource.Digest, func(w io.Writer) error { + return spillStream(ctx, stream, w, req.GetData(), maxBytes, s.log, req.resource.Digest) + }) if err != nil { return 0, 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} - } - - // 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) - } + // Closing the unlinked staging file is the whole cleanup: it frees the space. + defer s.closeStagingFile(f) 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, @@ -251,12 +193,6 @@ func (s *ByteStreamService) spillVerifyUpload(ctx context.Context, stream bytest 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 @@ -266,21 +202,11 @@ type backendUploadError struct{ err error } func (e *backendUploadError) Error() string { return e.err.Error() } func (e *backendUploadError) Unwrap() error { return e.err } -// 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 (e *digestMismatchError) Error() string { - return fmt.Sprintf("uploaded content does not match the declared digest: got=%s, want=%s", e.got, e.want) -} - // 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) { +// firstData is the payload already read from the first request. +func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, w io.Writer, firstData []byte, maxSize int64, log *log.Helper, digest string) error { + // running total, needed to enforce the cap before each write var size int64 write := func(data []byte) error { if len(data) == 0 { @@ -298,45 +224,47 @@ func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, // Write the data from the first request. if err := write(firstData); err != nil { - return size, err + return err } for { select { case <-ctx.Done(): // DeadlineExceeded, or Canceled - return size, ctx.Err() + return ctx.Err() default: // Extract the next chunk of data from the stream request req, err := getWriteRequest(stream) if err != nil { // Finished reading the stream is not a real error if errors.Is(err, io.EOF) { - return size, nil + return nil } - return size, err + return err } // 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 + return err } 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() { - return size, nil + return nil } } } } // Server-side streaming RPC for reading blobs, implements the bytestream interface -// NOTE: Due to the fact that we are using the OCI backend, we can not stream the content directly from the backend -// but instead we need to download the whole artifact and then stream it to the client +// NOTE: the content is not piped straight from the backend to the client. It is +// staged on disk and verified first (see stageDownload), so the first byte only +// leaves once the whole artifact is known to hash to the requested digest. The +// client therefore sees no data until the backend fetch has completed. func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.ByteStream_ReadServer) error { ctx := stream.Context() ctx, span := otelx.Start(ctx, byteStreamTracer, "ByteStreamService.Read") @@ -364,27 +292,25 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream. return sl.LogAndMaskErr(err, s.log) } - // streamwriter will stream chunks of data to the client - sw := &streamWriter{stream: stream, log: s.log, wantChecksum: req.ResourceName, gotChecksum: sha256.New()} - if err := backend.Download(ctx, sw, req.ResourceName); err != nil { - if isClientDisconnect(err) { - s.log.Infow("msg", "download canceled", "digest", req.ResourceName) - return nil - } - - return sl.LogAndMaskErr(err, s.log) + // Stage the backend egress on local disk and verify it against the requested + // digest BEFORE the first byte is sent: the client never receives content + // that does not hash to the key it asked for, and CAS memory stays bounded + // because the artifact lives on disk. + f, size, err := s.stageDownload(ctx, backend, req.ResourceName) + if err != nil { + return s.downloadError(err, req.ResourceName) } + defer s.closeStagingFile(f) - // check if the file has been tampered with and notify the client - if sw.GetChecksum() != req.ResourceName { - return kerrors.Unauthorized("checksum", fmt.Sprintf("checksum mismatch: got=%s, want=%s", sw.GetChecksum(), req.ResourceName)) + if err := copyStaged(sendWriter{stream}, f); err != nil { + return s.downloadError(err, req.ResourceName) } - s.log.Infow("msg", "download finished", "digest", req.ResourceName) + s.log.Infow("msg", "download finished", "digest", req.ResourceName, "size", size) s.audit.Dispatch(&events.CASArtifactDownloaded{ CASArtifactBase: &events.CASArtifactBase{ Digest: req.ResourceName, - SizeBytes: sw.size, + SizeBytes: size, BackendType: info.BackendType, }, }, info) @@ -392,6 +318,23 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream. return nil } +// downloadError maps a staging or streaming failure of a download to its gRPC +// status. A mismatch means the backend holds content that does not hash to its +// key, corrupt or tampered, so it is reported as DataLoss with both digests +// rather than blamed on the caller. A client that went away is not an error. +// Anything else is masked. +func (s *ByteStreamService) downloadError(err error, digest string) error { + if _, ok := errors.AsType[*digestMismatchError](err); ok { + return status.Error(codes.DataLoss, err.Error()) + } + if isClientDisconnect(err) { + s.log.Infow("msg", "download canceled", "digest", digest) + return nil + } + + return sl.LogAndMaskErr(err, s.log) +} + // checkUploadSize returns an ErrUploadSizeExceeded when total exceeds maxSize. // maxSize == 0 means no limit. func checkUploadSize(total, maxSize int64) error { @@ -439,32 +382,13 @@ func decodeResource(b64encoded string) (*v1.CASResource, error) { return resource, err } -// io.Writer wrapper for bytestreams.ReadResponses -type streamWriter struct { +// sendWriter adapts a bytestream Read stream to io.Writer: each Write becomes +// one ReadResponse. The content has already been verified on disk and the +// chunking is decided by the caller (copyStaged), so this is a pure adapter. +type sendWriter struct { stream bytestream.ByteStream_ReadServer - log *log.Helper - // expected wantChecksum of the data being sent - wantChecksum string - // calculated gotChecksum of the data sent - gotChecksum hash.Hash - // total number of bytes sent - size int64 } -// Send the chunk of data through the bytestream -func (sw *streamWriter) Write(data []byte) (int, error) { - sw.log.Debugw("msg", "sending download chunk", "digest", sw.wantChecksum, "chunkSize", len(data)) - - // Update the checksum of the data being sent - if _, err := sw.gotChecksum.Write(data); err != nil { - return 0, err - } - - sw.size += int64(len(data)) +func (sw sendWriter) Write(data []byte) (int, error) { return len(data), sw.stream.Send(&bytestream.ReadResponse{Data: data}) } - -// GetChecksum retrieves the sha256 checksum of the read contents -func (sw *streamWriter) GetChecksum() string { - return hex.EncodeToString(sw.gotChecksum.Sum(nil)) -} diff --git a/app/artifact-cas/internal/service/bytestream_download_test.go b/app/artifact-cas/internal/service/bytestream_download_test.go index 3025cab29..3d15054a7 100644 --- a/app/artifact-cas/internal/service/bytestream_download_test.go +++ b/app/artifact-cas/internal/service/bytestream_download_test.go @@ -17,13 +17,12 @@ package service import ( "context" - "crypto/sha256" "errors" "fmt" "io" + "os" "testing" - "github.com/chainloop-dev/chainloop/pkg/servicelogger" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -32,12 +31,12 @@ import ( ) // These tests lock down the DOWNLOAD digest-verification behavior — the server -// must stream the stored bytes back, compute their sha256 across however many +// must stage the stored bytes on disk, compute their sha256 across however many // chunks the backend produces, and reject any content whose digest does not -// match the requested resource name. +// match the requested resource name BEFORE the first byte reaches the client. // fakeReadServer is a minimal bytestream.ByteStream_ReadServer that records the -// data chunks the streamWriter sends. Only Send is exercised by streamWriter. +// data chunks the sendWriter sends. Only Send is exercised by sendWriter. type fakeReadServer struct { bytestream.ByteStream_ReadServer sent [][]byte @@ -51,36 +50,23 @@ func (f *fakeReadServer) Send(r *bytestream.ReadResponse) error { return nil } -// TestStreamWriter_ChecksumAcrossWrites verifies the download writer computes -// the sha256 over the full stream regardless of how the content is chunked, and -// forwards every byte to the client in order. -func TestStreamWriter_ChecksumAcrossWrites(t *testing.T) { +// TestSendWriter_ForwardsChunksInOrder verifies the download adapter turns +// every Write into one ReadResponse carrying exactly those bytes, in order. +// Hashing is not its job: content is verified on disk before the first Send. +func TestSendWriter_ForwardsChunksInOrder(t *testing.T) { content := []byte("chainloop download payload delivered in several writes") - digest := sha256Hex(content) fake := &fakeReadServer{} - sw := &streamWriter{ - stream: fake, - log: servicelogger.EmptyLogger(), - wantChecksum: digest, - gotChecksum: sha256.New(), - } + sw := sendWriter{stream: fake} - // Feed the content in uneven pieces to mimic a backend emitting arbitrary - // chunk boundaries. + // Feed the content in uneven pieces to mimic arbitrary chunk boundaries. for off := 0; off < len(content); off += 11 { - end := off + 11 - if end > len(content) { - end = len(content) - } + end := min(off+11, len(content)) n, err := sw.Write(content[off:end]) require.NoError(t, err) assert.Equal(t, end-off, n) } - assert.Equal(t, digest, sw.GetChecksum(), "checksum must be computed over the whole stream") - assert.Equal(t, int64(len(content)), sw.size) - // The bytes forwarded to the client must reassemble to the exact content. forwarded := make([]byte, 0, len(content)) for _, c := range fake.sent { @@ -89,14 +75,6 @@ func TestStreamWriter_ChecksumAcrossWrites(t *testing.T) { assert.Equal(t, content, forwarded) } -// TestStreamWriter_EmptyStreamChecksum: with no writes the computed checksum is -// the sha256 of the empty input. -func TestStreamWriter_EmptyStreamChecksum(t *testing.T) { - sw := &streamWriter{stream: &fakeReadServer{}, log: servicelogger.EmptyLogger(), gotChecksum: sha256.New()} - assert.Equal(t, sha256Hex([]byte{}), sw.GetChecksum()) - assert.Equal(t, int64(0), sw.size) -} - // recvAllDownload drains a Read stream, returning the concatenated payload and // the terminating error (io.EOF becomes nil). func recvAllDownload(reader bytestream.ByteStream_ReadClient) ([]byte, error) { @@ -171,9 +149,10 @@ func (s *bytestreamSuite) TestDownloadEmptyContentChecksum() { s.Equal(int64(0), info.SizeBytes) } -// TestDownloadTamperedAcrossChunksRejected: if the streamed bytes do not hash to -// the requested digest, the server reports a checksum mismatch and emits no -// audit event — the tamper-detection guarantee of a CAS. +// TestDownloadTamperedAcrossChunksRejected: if the stored bytes do not hash to +// the requested digest, the server reports the content as lost/corrupt, sends +// NOT A SINGLE BYTE to the client and emits no audit event — the +// tamper-detection guarantee of a CAS. func (s *bytestreamSuite) TestDownloadTamperedAcrossChunksRejected() { // The client asks for this digest, but the backend returns different bytes. requested := sha256Hex([]byte("the authentic artifact contents")) @@ -189,12 +168,33 @@ func (s *bytestreamSuite) TestDownloadTamperedAcrossChunksRejected() { reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: requested}) s.NoError(err) - _, err = recvAllDownload(reader) - s.ErrorContains(err, "checksum mismatch") + got, err := recvAllDownload(reader) + assertGRPCError(s.T(), err, codes.DataLoss, "does not match the requested digest") + s.ErrorContains(err, "got="+sha256Hex(tampered)) + s.Empty(got, "no unverified byte may reach the client") // tampered downloads emit no events s.Empty(s.audit.published) } +// TestDownloadStagingFailureMasked: when the download cannot be staged on disk +// (here: the staging directory vanished) the failure is masked as Internal, +// nothing is sent to the client and no audit event is emitted. +func (s *bytestreamSuite) TestDownloadStagingFailureMasked() { + content := []byte("never leaves the server") + digest := sha256Hex(content) + s.Require().NoError(os.RemoveAll(s.stagingDir)) + + reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: digest}) + s.NoError(err) + + got, err := recvAllDownload(reader) + assertGRPCError(s.T(), err, codes.Internal, "server error") + s.Empty(got) + s.Empty(s.audit.published) + // the backend is never asked for content that cannot be staged + s.ociBackend.AssertNotCalled(s.T(), "Download", mock.Anything, mock.Anything, mock.Anything) +} + // TestDownloadClientDisconnect: a backend Download failure that indicates the // client went away is treated as a cancellation (no error to a gone client, no // audit). This is pre-existing download behavior; the test pins it. diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go index 8fc6d0adb..9eda0b4db 100644 --- a/app/artifact-cas/internal/service/bytestream_test.go +++ b/app/artifact-cas/internal/service/bytestream_test.go @@ -400,13 +400,10 @@ func (s *bytestreamSuite) TestDownloadFoundMistmathedDigest() { reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: "deadbeef"}) s.NoError(err) - // receive the data, it should contain all of it since the buffer is serverside is 1MB + // The content is verified on disk before anything is sent, so the very first + // Recv reports the mismatch and no data ever reaches the client. got, err := reader.Recv() - s.NoError(err) - s.Equal("hello world", string(got.Data)) - // Return a mistmached digest - got, err = reader.Recv() - s.ErrorContains(err, "checksum mismatch:") + assertGRPCError(s.T(), err, codes.DataLoss, "does not match the requested digest") s.Nil(got) // tampered downloads emit no events s.Empty(s.audit.published) @@ -435,8 +432,9 @@ type bytestreamSuite struct { 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 is the per-test staging directory (uploads and downloads) the + // service is configured with, so tests can assert it is left clean or make + // it unavailable. stagingDir string } diff --git a/app/artifact-cas/internal/service/download.go b/app/artifact-cas/internal/service/download.go index dcc6f8f34..5264b4e4f 100644 --- a/app/artifact-cas/internal/service/download.go +++ b/app/artifact-cas/internal/service/download.go @@ -16,9 +16,7 @@ package service import ( - "bytes" - "crypto/sha256" - "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -101,58 +99,55 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) { if filename == "" { filename = info.FileName } - // if the buffer contains the actual data we expect we proceed with sending it to the browser - // Set headers - w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%s", filename)) - w.Header().Set("Content-Length", strconv.FormatInt(info.Size, 10)) s.log.Infow("msg", "download initialized", "digest", wantChecksum, "size", bytefmt.ByteSize(uint64(info.Size))) - gotChecksum := sha256.New() - // create temporary buffer to write to both the writer and the checksum - buf := bytes.NewBuffer(nil) - - // NOTE: we don't sent the file directly to the writer because we need to calculate the checksum - // and we want to send the file / even if partially only if the checksum matches - // this has a performance impact but it's the only way to ensure that the file is not corrupted - // and don't require client-side verification - mw := io.MultiWriter(buf, gotChecksum) - if err := b.Download(ctx, mw, wantChecksum.Hex); err != nil { - if isClientDisconnect(err) { - s.log.Infow("msg", "download canceled", "digest", wantChecksum) - return - } - - http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError) + // Stage the backend egress on local disk and verify it against the requested + // digest before anything is written to the response: the browser never + // receives content that does not hash to the digest it asked for, and CAS + // memory stays bounded because the artifact lives on disk, not in a buffer. + f, size, err := s.stageDownload(ctx, b, wantChecksum.Hex) + if err != nil { + s.writeDownloadError(w, err, wantChecksum.Hex) return } + defer s.closeStagingFile(f) + + // The content is verified: announce it to the browser with its exact size + w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%s", filename)) + w.Header().Set("Content-Length", strconv.FormatInt(size, 10)) - // Verify the checksum - if got, want := hex.EncodeToString(gotChecksum.Sum(nil)), wantChecksum.Hex; got != want { - msg := fmt.Sprintf("checksums mismatch: got: %s, want: %s", got, want) - s.log.Info(msg) - http.Error(w, msg, http.StatusUnauthorized) + // A plain io.Copy lets the response writer pull the file with sendfile, so + // the verified bytes go kernel-to-kernel without a user-space buffer. + if _, err := io.Copy(w, f); err != nil { + s.writeDownloadError(w, err, wantChecksum.Hex) return } - // the backend egress completed and the content was verified, record the download + s.log.Infow("msg", "download finished", "digest", wantChecksum, "size", bytefmt.ByteSize(uint64(size))) s.audit.Dispatch(&events.CASArtifactDownloaded{ CASArtifactBase: &events.CASArtifactBase{ Digest: wantChecksum.Hex, - SizeBytes: info.Size, + SizeBytes: size, FileName: filename, BackendType: auth.BackendType, }, }, auth) +} - if _, err := io.Copy(w, buf); err != nil { - if isClientDisconnect(err) { - s.log.Infow("msg", "download canceled during response write", "digest", wantChecksum) - return - } - - http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError) +// writeDownloadError maps a staging or streaming failure of a download to its +// HTTP response. A mismatch means the backend holds content that does not hash +// to its key, corrupt or tampered, so it is a 500 carrying both digests rather +// than blamed on the caller. A client that went away gets nothing. Anything +// else is masked. +func (s *DownloadService) writeDownloadError(w http.ResponseWriter, err error, digest string) { + if _, ok := errors.AsType[*digestMismatchError](err); ok { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if isClientDisconnect(err) { + s.log.Infow("msg", "download canceled", "digest", digest) return } - s.log.Infow("msg", "download finished", "digest", wantChecksum, "size", bytefmt.ByteSize(uint64(info.Size))) + http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError) } diff --git a/app/artifact-cas/internal/service/download_test.go b/app/artifact-cas/internal/service/download_test.go index c6a219a89..49324c736 100644 --- a/app/artifact-cas/internal/service/download_test.go +++ b/app/artifact-cas/internal/service/download_test.go @@ -19,6 +19,8 @@ import ( "io" "net/http" "net/http/httptest" + "os" + "strconv" "testing" v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" @@ -51,11 +53,15 @@ func TestDownloadServiceAuditEvents(t *testing.T) { } tests := []struct { - name string - content string - claims *casJWT.Claims - wantStatus int - wantEvents int + name string + content string + claims *casJWT.Claims + // removeStagingDir makes staging impossible before the request is served + removeStagingDir bool + wantStatus int + // wantBodyContains is asserted on the response body when non-empty + wantBodyContains string + wantEvents int }{ { name: "successful download emits an event", @@ -65,10 +71,19 @@ func TestDownloadServiceAuditEvents(t *testing.T) { wantEvents: 1, }, { - name: "checksum mismatch emits no event", - content: "tampered content", - claims: downloaderClaims(false), - wantStatus: http.StatusUnauthorized, + name: "checksum mismatch is reported as corrupt content, sends no bytes and emits no event", + content: "tampered content", + claims: downloaderClaims(false), + wantStatus: http.StatusInternalServerError, + wantBodyContains: "does not match the requested digest", + }, + { + name: "staging failure is masked, sends no bytes and emits no event", + content: "hello world", + claims: downloaderClaims(false), + removeStagingDir: true, + wantStatus: http.StatusInternalServerError, + wantBodyContains: "server error", }, { name: "internal control plane traffic emits no event", @@ -86,17 +101,25 @@ func TestDownloadServiceAuditEvents(t *testing.T) { uploaderDownloader.On("Describe", mock.Anything, digestHex).Return(&v1.CASResource{ FileName: "test.txt", Digest: digestHex, Size: int64(len(tc.content)), }, nil) - uploaderDownloader.On("Download", mock.Anything, mock.Anything, digestHex).Return(nil). - Run(func(args mock.Arguments) { - _, err := io.WriteString(args.Get(1).(io.Writer), tc.content) - require.NoError(t, err) - }) + if !tc.removeStagingDir { + uploaderDownloader.On("Download", mock.Anything, mock.Anything, digestHex).Return(nil). + Run(func(args mock.Arguments) { + _, err := io.WriteString(args.Get(1).(io.Writer), tc.content) + require.NoError(t, err) + }) + } + + stagingDir := t.TempDir() + if tc.removeStagingDir { + require.NoError(t, os.RemoveAll(stagingDir)) + } audit := &fakePublisher{} svc := NewDownloadService( backend.Providers{backendType: provider}, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(audit)), + WithStagingDir(stagingDir), ) req := httptest.NewRequest(http.MethodGet, "/download/sha256:"+digestHex, nil) @@ -107,6 +130,20 @@ func TestDownloadServiceAuditEvents(t *testing.T) { svc.ServeHTTP(w, req) assert.Equal(t, tc.wantStatus, w.Code) + if tc.wantStatus == http.StatusOK { + assert.Equal(t, tc.content, w.Body.String()) + assert.Equal(t, strconv.Itoa(len(tc.content)), w.Header().Get("Content-Length")) + assert.Equal(t, "attachment; filename=test.txt", w.Header().Get("Content-Disposition")) + } else { + assert.NotContains(t, w.Body.String(), tc.content, "no unverified byte may reach the client") + assert.Contains(t, w.Body.String(), tc.wantBodyContains) + } + if !tc.removeStagingDir { + // the staging dir is left clean on every exit path + entries, err := os.ReadDir(stagingDir) + require.NoError(t, err) + assert.Empty(t, entries) + } require.Len(t, audit.published, tc.wantEvents) if tc.wantEvents == 0 { return diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index 3f055c3c9..8aff6e466 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -38,10 +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 is the local directory where uploads and downloads are staged + // on disk while their SHA256 is verified: uploads before reaching the + // backend, downloads before the first byte is sent to the client. It must be + // set and writable: leaving it empty is a deployment error and transfers are + // refused rather than staged somewhere unintended. stagingDir string } @@ -77,10 +78,10 @@ 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. +// WithStagingDir sets the local directory where uploads and downloads are +// spilled and verified before being forwarded. It must point at a writable +// volume dedicated to this pod; there is no default, so a transfer fails loudly +// rather than silently staging somewhere unintended. func WithStagingDir(dir string) NewOpt { return func(s *commonService) { s.stagingDir = dir diff --git a/app/artifact-cas/internal/service/staging.go b/app/artifact-cas/internal/service/staging.go new file mode 100644 index 000000000..93d180733 --- /dev/null +++ b/app/artifact-cas/internal/service/staging.go @@ -0,0 +1,187 @@ +// +// 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" + "errors" + "fmt" + "io" + "os" + + backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" +) + +// stagingFilePrefix names the per-request temporary files, followed by the +// direction. 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 request appears as +// "/cas-staging-upload-NNN (deleted)". +const stagingFilePrefix = "cas-staging-" + +// stagingKind is the direction of a staged transfer. It names the staging file +// and phrases a digest mismatch for the party that can act on it. +type stagingKind struct { + name string + // mismatchMsg explains whose digest the content failed to match + mismatchMsg string +} + +var ( + // stagingUpload: the client declared a digest that does not describe the + // bytes it sent. The request is invalid. + stagingUpload = stagingKind{name: "upload", mismatchMsg: "uploaded content does not match the declared digest"} + // stagingDownload: the backend returned bytes that do not hash to the key + // they are stored under. The stored content is corrupt or tampered. + stagingDownload = stagingKind{name: "download", mismatchMsg: "stored content does not match the requested digest"} +) + +// stageAndVerify spills whatever fill writes into an unlinked temporary file +// under the staging directory, hashing the bytes in the same pass, and compares +// the result with wantDigest (a hex-encoded SHA256). It is the single integrity +// gate shared by uploads (the producer is the client stream) and downloads (the +// producer is the storage backend), so the paths cannot drift. +// +// On success it returns the file rewound to offset 0 and the number of bytes +// staged; the caller owns the file and must release it with closeStagingFile, +// which frees the space. On any failure the file is already closed and nil is +// returned: a digest mismatch as a *digestMismatchError, a producer failure +// unwrapped for the caller to classify, and staging-disk failures wrapped with +// context. +// +// Staging on disk rather than in memory keeps this service's memory use +// independent of artifact size. +func (s *commonService) stageAndVerify(kind stagingKind, wantDigest string, fill func(w io.Writer) error) (*os.File, 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 nil, 0, errors.New("no staging directory configured") + } + + f, err := os.CreateTemp(s.stagingDir, stagingFilePrefix+kind.name+"-*") + if err != nil { + return nil, 0, fmt.Errorf("creating staging file: %w", err) + } + + // From here on, closing the file is the whole cleanup on every exit path. + fail := func(err error) (*os.File, int64, error) { + s.closeStagingFile(f) + return nil, 0, 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-request 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 request 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 { + return fail(fmt.Errorf("unlinking staging file: %w", err)) + } + + // Tee the producer into the file and a SHA256 hasher in one pass. + hasher := sha256.New() + if err := fill(io.MultiWriter(f, hasher)); err != nil { + return fail(err) + } + + // Fail closed: if the bytes do not hash to the expected digest, the file is + // discarded and nothing downstream ever sees it. + if got := hex.EncodeToString(hasher.Sum(nil)); got != wantDigest { + return fail(&digestMismatchError{kind: kind, got: got, want: wantDigest}) + } + + // The write offset is the staged size. Then rewind so the consumer 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. + size, err := f.Seek(0, io.SeekCurrent) + if err != nil { + return fail(fmt.Errorf("sizing staging file: %w", err)) + } + if _, err := f.Seek(0, io.SeekStart); err != nil { + return fail(fmt.Errorf("rewinding staging file: %w", err)) + } + + return f, size, nil +} + +// stageDownload stages the backend's copy of digest and verifies it. It is the +// download half of stageAndVerify shared by the gRPC and HTTP handlers, which +// keep only their transport-specific status mapping. A mismatch is logged here, +// once, with the digest the backend actually returned so the operator can +// investigate the backend, and returned as a *digestMismatchError. +func (s *commonService) stageDownload(ctx context.Context, b backend.Downloader, digest string) (*os.File, int64, error) { + f, size, err := s.stageAndVerify(stagingDownload, digest, func(w io.Writer) error { + return b.Download(ctx, w, digest) + }) + if err != nil { + if mismatch, ok := errors.AsType[*digestMismatchError](err); ok { + s.log.Warnw("msg", "download rejected: "+stagingDownload.mismatchMsg, "digest", digest, "got", mismatch.got) + } + return nil, 0, err + } + + s.log.Infow("msg", "artifact verified, streaming to client", "digest", digest, "size", size) + return f, size, nil +} + +// closeStagingFile releases a file returned by stageAndVerify. The file is +// already unlinked, so closing it is what frees the space on the staging volume. +func (s *commonService) closeStagingFile(f *os.File) { + if err := f.Close(); err != nil { + s.log.Warnw("msg", "failed to close staging file", "error", err.Error()) + } +} + +// stagedChunkSize is the size of the gRPC ReadResponse messages a verified +// staging file is streamed in. It stays under grpc-go's largest pooled buffer +// tier (1 MiB) once proto framing is added, so each Send reuses a pooled buffer +// instead of allocating a fresh one. +const stagedChunkSize = 512 << 10 + +// copyStaged streams a verified staging file to the gRPC client through w in +// stagedChunkSize pieces. The file is deliberately wrapped so its io.WriterTo is +// not picked up by io.CopyBuffer, which would otherwise decide the chunking (and +// therefore the gRPC message size) on its own. The HTTP handler does not use +// this: a plain io.Copy lets the response take the kernel sendfile path. +func copyStaged(w io.Writer, f *os.File) error { + _, err := io.CopyBuffer(w, struct{ io.Reader }{f}, make([]byte, stagedChunkSize)) + return err +} + +// digestMismatchError marks content whose bytes do not hash to the expected +// digest. The kind says whose digest was wrong: on upload the client's declared +// one, on download the backend's stored content. Callers map it to the status +// that fits their direction; no unverified byte is ever forwarded either way. +type digestMismatchError struct { + kind stagingKind + got, want string +} + +func (e *digestMismatchError) Error() string { + return fmt.Sprintf("%s: got=%s, want=%s", e.kind.mismatchMsg, e.got, e.want) +} diff --git a/app/artifact-cas/internal/service/staging_test.go b/app/artifact-cas/internal/service/staging_test.go new file mode 100644 index 000000000..f10110fc7 --- /dev/null +++ b/app/artifact-cas/internal/service/staging_test.go @@ -0,0 +1,152 @@ +// +// 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 ( + "errors" + "io" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests lock down the shared spill-and-verify primitive used by uploads +// and both download paths: whatever the producer writes is spilled to an +// unlinked file under the staging directory, hashed in the same pass and +// compared with the expected digest. Only a verified file is ever handed back. + +func TestStageAndVerify(t *testing.T) { + content := []byte("bytes that are spilled to disk, hashed and verified") + digest := sha256Hex(content) + tampered := []byte("tampered") + errProducer := errors.New("producer failed") + + tests := []struct { + name string + // stagingDir overrides the per-test temp dir when set + stagingDir func(t *testing.T) string + want string + fill func(w io.Writer) error + // wantErr is matched with errors.Is, wantErrMsg with ErrorContains + wantErr error + wantErrMsg string + }{ + { + name: "matching content is returned rewound with its size", + want: digest, + fill: func(w io.Writer) error { + // uneven chunks: the hash must cover the whole stream + _, _ = w.Write(content[:7]) + _, err := w.Write(content[7:]) + return err + }, + }, + { + name: "empty content verifies against sha256 of the empty input", + want: sha256Hex([]byte{}), + fill: func(_ io.Writer) error { return nil }, + }, + { + name: "content that does not hash to the digest is rejected with both digests", + want: digest, + fill: func(w io.Writer) error { + _, err := w.Write(tampered) + return err + }, + wantErrMsg: stagingUpload.mismatchMsg + ": got=" + sha256Hex(tampered) + ", want=" + digest, + }, + { + name: "producer error is returned unwrapped for the caller to classify", + want: digest, + fill: func(_ io.Writer) error { return errProducer }, + wantErr: errProducer, + }, + { + name: "an unconfigured staging directory is refused", + stagingDir: func(*testing.T) string { return "" }, + want: digest, + fill: func(_ io.Writer) error { return nil }, + wantErrMsg: "no staging directory configured", + }, + { + name: "a missing staging directory fails to create the file", + stagingDir: func(t *testing.T) string { return filepath.Join(t.TempDir(), "does-not-exist") }, + want: digest, + fill: func(_ io.Writer) error { return nil }, + wantErrMsg: "creating staging file", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + svc := newCommonService(nil, WithStagingDir(dir)) + if tc.stagingDir != nil { + svc.stagingDir = tc.stagingDir(t) + } + + f, size, err := svc.stageAndVerify(stagingUpload, tc.want, tc.fill) + + // whatever happened, nothing may be left behind in the staging dir + entries, readErr := os.ReadDir(dir) + require.NoError(t, readErr) + assert.Empty(t, entries, "staging dir must be left clean") + + if tc.wantErr != nil || tc.wantErrMsg != "" { + assert.Nil(t, f) + if tc.wantErr != nil { + require.ErrorIs(t, err, tc.wantErr) + } + if tc.wantErrMsg != "" { + require.ErrorContains(t, err, tc.wantErrMsg) + } + return + } + + require.NoError(t, err) + require.NotNil(t, f) + defer f.Close() + + got, err := io.ReadAll(f) + require.NoError(t, err) + assert.Equal(t, int64(len(got)), size) + assert.Equal(t, tc.want, sha256Hex(got), "file must be rewound and hold the verified bytes") + }) + } +} + +// TestStageAndVerify_FileIsUnlinked: the staged file is removed from the +// directory while it is still open, so a crash mid-stream cannot strand it. +func TestStageAndVerify_FileIsUnlinked(t *testing.T) { + dir := t.TempDir() + svc := newCommonService(nil, WithStagingDir(dir)) + + var seenDuringFill []os.DirEntry + f, _, err := svc.stageAndVerify(stagingUpload, sha256Hex([]byte("x")), func(w io.Writer) error { + var readErr error + seenDuringFill, readErr = os.ReadDir(dir) + require.NoError(t, readErr) + _, err := w.Write([]byte("x")) + return err + }) + require.NoError(t, err) + defer f.Close() + + assert.Empty(t, seenDuringFill, "the staging file must be unlinked before the producer runs") +} diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index 3491d8a30..52820a3a5 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.443.0 +version: 1.443.1 # Do not update appVersion, this is handled automatically by the release process appVersion: v1.110.0 diff --git a/deployment/chainloop/templates/_helpers.tpl b/deployment/chainloop/templates/_helpers.tpl index 4665545e0..e530766fd 100644 --- a/deployment/chainloop/templates/_helpers.tpl +++ b/deployment/chainloop/templates/_helpers.tpl @@ -494,10 +494,10 @@ Return the Nats connection string for the CAS {{- 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. +Directory where the CAS stages and verifies uploads and downloads before +forwarding them. 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 diff --git a/deployment/chainloop/templates/cas/configmap.yaml b/deployment/chainloop/templates/cas/configmap.yaml index 402bef8fb..fd0aa6e63 100644 --- a/deployment/chainloop/templates/cas/configmap.yaml +++ b/deployment/chainloop/templates/cas/configmap.yaml @@ -14,8 +14,8 @@ 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. + # Local directory where uploads and downloads are staged and verified against + # their digest before being forwarded. Backed by the staging emptyDir. staging_dir: {{ include "chainloop.cas.staging_dir" . | quote }} server: http: diff --git a/deployment/chainloop/templates/cas/deployment.yaml b/deployment/chainloop/templates/cas/deployment.yaml index b99892cc2..b55a12e7d 100644 --- a/deployment/chainloop/templates/cas/deployment.yaml +++ b/deployment/chainloop/templates/cas/deployment.yaml @@ -126,8 +126,8 @@ 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). + # Writable scratch volume for staging + verifying uploads and downloads + # before they are forwarded (the container root filesystem is read-only). - name: staging mountPath: {{ include "chainloop.cas.staging_dir" . | quote }} {{- if eq "gcpSecretManager" .Values.secretsBackend.backend }} @@ -162,9 +162,9 @@ 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. + # Node-disk (NOT tmpfs/RAM) scratch space for staging + verifying uploads + # and downloads. No sizeLimit: breaching one evicts the Pod rather than + # failing the transfer that overran it. - name: staging emptyDir: {} {{- if include "cas.tls-secret-name" . }} diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index e3e47380b..f05d413b0 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.yaml @@ -1479,19 +1479,22 @@ cas: 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. + ## CAS staging. + ## Uploads and downloads are streamed to a directory on local disk and verified + ## against their digest before being forwarded: uploads before they reach the + ## storage backend, downloads before the first byte is sent to the client. + ## Nothing unverified ever leaves the CAS in either direction. 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 + ## kubelet instead of failing the offending transfer. Bound uploads with the ## per-request size cap, and set cas.resources ephemeral-storage requests to - ## have the scheduler account for the space. + ## have the scheduler account for the space. Size for the number of concurrent + ## uploads AND downloads you expect times the largest artifact. ## @param cas.automountServiceAccountToken Mount Service Account token in cas pods ##