feat(artifact-cas): verify uploaded content against the declared digest

The Artifact CAS keys stored artifacts by their client-declared SHA256. This
change adds end-to-end content-integrity verification to the upload path so the
canonical key is only ever written from bytes CAS has confirmed hash to that
digest, across all object-store backends (S3, S3 access point, Azure Blob), in
addition to the OCI backend's existing layer-digest check.

The CAS service streams each upload to a temporary file on a local staging
volume, computes its SHA256 while receiving, and verifies it against the declared
digest before handing the verified file to the backend. A mismatch is rejected
with InvalidArgument and nothing is written to the backend. The upload path is
consolidated into a single spill-then-upload flow shared by all backends,
replacing the per-backend streaming branch and the StreamingUploader interface.

The staging directory is configured through staging_dir and is required: the CAS
refuses to start without one, and proves it is writable before serving traffic,
so a misconfiguration surfaces at boot rather than as a healthy service that
fails every upload. The Helm chart mounts a dedicated emptyDir for it. Every exit
path of the upload handler removes its own staging file, so the staging volume is
not left holding content once a request ends.

Assisted-by: Claude Code
Signed-off-by: Javier Rodriguez <javier@chainloop.dev>

Chainloop-Trace-Sessions: 5d0fc605-25b0-4bb3-900c-2c2a8fb4922f

Author: javirln

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)

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")
+		})
+	}
+}

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

app/artifact-cas/cmd/wire_gen.go

@@ -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}

app/artifact-cas/configs/config.devel.yaml

@@ -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"

app/artifact-cas/internal/conf/conf.pb.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