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

app/artifact-cas/cmd/main.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf"
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server"
+	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service"
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/chainloop-dev/chainloop/pkg/credentials"
 	"github.com/chainloop-dev/chainloop/pkg/credentials/manager"
@@ -121,6 +122,12 @@ func main() {
 
 	_ = logger.Log(log.LevelInfo, "msg", "starting artifact-cas service", "version", Version)
 
+	// Ensure the upload staging directory exists and sweep any files a previous
+	// crash left behind, so verification always starts from a clean volume.
+	if err := prepareStagingDir(&bc, logger); err != nil {
+ panic(err)
+	}
+
 	flush, err := initSentry(&bc, logger)
 	defer flush()
 	if err != nil {
@@ -152,6 +159,31 @@ func newProtoValidator() (protovalidate.Validator, error) {
 	return protovalidate.New()
 }
 
+// prepareStagingDir resolves the upload staging directory (falling back to the
+// OS temp dir when unconfigured — dev only; production mounts a dedicated
+// volume), creates it, and removes any leftover staging files from a previous
+// run. It must use the same directory the service is configured with (see
+// serviceOpts / conf.staging_dir).
+func prepareStagingDir(bc *conf.Bootstrap, logger log.Logger) error {
+	dir := bc.GetStagingDir()
+	if dir == "" {
+ dir = os.TempDir()
+ _ = logger.Log(log.LevelWarn, "msg", "staging_dir not configured, falling back to OS temp dir (dev only)", "dir", dir)
+	}
+
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+ return err
+	}
+
+	if _, err := service.SweepStagingDir(dir, servicelogger.ScopedHelper(logger, "staging")); err != nil {
+ // A sweep failure is not fatal: the deferred per-upload cleanup still
+ // applies, so log and continue rather than blocking startup.
+ _ = logger.Log(log.LevelWarn, "msg", "failed to sweep staging dir", "dir", dir, "error", err.Error())
+	}
+
+	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/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/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"