feat(cas): Azure Blob Storage support (#360)

Signed-off-by: Miguel Martinez Trivino <[email protected]>

Author: migmartri

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"
 	backend "github.com/chainloop-dev/chainloop/internal/blobmanager"
+	"github.com/chainloop-dev/chainloop/internal/blobmanager/azureblob"
 	"github.com/chainloop-dev/chainloop/internal/blobmanager/oci"
 	"github.com/chainloop-dev/chainloop/internal/credentials"
 	"github.com/chainloop-dev/chainloop/internal/credentials/manager"
@@ -64,11 +65,12 @@ type app struct {
 }
 
 func loadCASBackendProviders(creader credentials.Reader) backend.Providers {
-	// Currently only OCI is supported
-	// Here we will load the rest of providers, S3, GCS, etc
-	p := oci.NewBackendProvider(creader)
+	// Initialize CAS backend providers
+	ociProvider := oci.NewBackendProvider(creader)
+	azureBlobProvider := azureblob.NewBackendProvider(creader)
 	return backend.Providers{
-		p.ID(): p,
+		ociProvider.ID():       ociProvider,
+		azureBlobProvider.ID(): azureBlobProvider,
 	}
 }
 

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

@@ -5,10 +5,12 @@
 server:
   http:
     addr: 0.0.0.0:8001
-    timeout: 1s
+    timeout: 5s
   grpc:
     addr: 0.0.0.0:9001
-    timeout: 1s
+    # Some cas backends are slow, so we need to increase the timeout
+    # for example, Azure Blob Storage describe takes more than 1 second to respond sometimes
+    timeout: 5s
   http_metrics:
     addr: 0.0.0.0:5001
 

app/artifact-cas/internal/service/bytestream.go

@@ -18,8 +18,11 @@ package service
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
 	"encoding/base64"
 	"encoding/gob"
+	"fmt"
+	"hash"
 	"io"
 
 	"errors"
@@ -113,7 +116,7 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 		return sl.LogAndMaskErr(err, s.log)
 	}
 
-	s.log.Infow("msg", "artifact received, uploading now to OCI backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", buffer.size)
+	s.log.Infow("msg", "artifact received, uploading now to backend", "name", req.resource.FileName, "digest", req.resource.Digest, "size", buffer.size)
 	if err := backend.Upload(ctx, buffer, req.resource); err != nil {
 		return sl.LogAndMaskErr(err, s.log)
 	}
@@ -151,7 +154,7 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.
 	}
 
 	// streamwriter will stream chunks of data to the client
-	sw := &streamWriter{stream, s.log, req.ResourceName}
+	sw := &streamWriter{stream, s.log, req.ResourceName, sha256.New()}
 	if err := backend.Download(ctx, sw, req.ResourceName); err != nil {
 		if errors.Is(err, context.Canceled) {
 			s.log.Infow("msg", "download canceled", "digest", req.ResourceName)
@@ -161,6 +164,11 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.
 		return sl.LogAndMaskErr(err, s.log)
 	}
 
+	// 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))
+	}
+
 	s.log.Infow("msg", "download finished", "digest", req.ResourceName)
 
 	return nil
@@ -269,11 +277,24 @@ func decodeResource(b64encoded string) (*v1.CASResource, error) {
 type streamWriter struct {
 	stream bytestream.ByteStream_ReadServer
 	log    *log.Helper
-	digest string
+	// expected wantChecksum of the data being sent
+	wantChecksum string
+	// calculated gotChecksum of the data sent
+	gotChecksum hash.Hash
 }
 
 // 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.digest, "chunkSize", len(data))
+	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
+	}
 	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 fmt.Sprintf("%x", sw.gotChecksum.Sum(nil))
+}

app/artifact-cas/internal/service/bytestream_test.go

@@ -201,6 +201,27 @@ func (s *bytestreamSuite) TestReadErrorDownloading() {
 }
 
 func (s *bytestreamSuite) TestDownloadOk() {
+	s.ociBackend.On("Download", mock.Anything, mock.Anything, "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9").
+		Return(nil).Run(func(args mock.Arguments) {
+		buf := bytes.NewBuffer([]byte("hello world"))
+		_, err := io.Copy(args.Get(1).(io.Writer), buf)
+		s.NoError(err)
+	})
+
+	reader, err := s.client.Read(s.downCtx, &bytestream.ReadRequest{ResourceName: "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"})
+	s.NoError(err)
+
+	// receive the data, it should contain all of it since the buffer is server side is 1MB
+	got, err := reader.Recv()
+	s.NoError(err)
+	s.Equal("hello world", string(got.Data))
+	// EOF
+	got, err = reader.Recv()
+	s.ErrorIs(err, io.EOF)
+	s.Nil(got)
+}
+
+func (s *bytestreamSuite) TestDownloadFoundMistmathedDigest() {
 	s.ociBackend.On("Download", mock.Anything, mock.Anything, "deadbeef").
 		Return(nil).Run(func(args mock.Arguments) {
 		buf := bytes.NewBuffer([]byte("hello world"))
@@ -215,9 +236,9 @@ func (s *bytestreamSuite) TestDownloadOk() {
 	got, err := reader.Recv()
 	s.NoError(err)
 	s.Equal("hello world", string(got.Data))
-	// EOF
+	// Return a mistmached digest
 	got, err = reader.Recv()
-	s.ErrorIs(err, io.EOF)
+	s.ErrorContains(err, "checksum mismatch:")
 	s.Nil(got)
 }
 

app/artifact-cas/internal/service/download.go

@@ -16,9 +16,12 @@
 package service
 
 import (
+	"bytes"
 	"context"
+	"crypto/sha256"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strconv"
 
@@ -58,7 +61,7 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	hash, err := cr_v1.NewHash(digest)
+	wantChecksum, err := cr_v1.NewHash(digest)
 	if err != nil {
 		http.Error(w, "invalid digest", http.StatusBadRequest)
 		return
@@ -79,7 +82,7 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	info, err := b.Describe(ctx, hash.Hex)
+	info, err := b.Describe(ctx, wantChecksum.Hex)
 	if err != nil && backend.IsNotFound(err) {
 		http.Error(w, "artifact not found", http.StatusNotFound)
 		return
@@ -88,20 +91,44 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Set headers
-	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%s", info.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)))
 
-	s.log.Infow("msg", "download initialized", "digest", hash, "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)
 
-	if err := b.Download(ctx, w, hash.Hex); err != 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 errors.Is(err, context.Canceled) {
-			s.log.Infow("msg", "download canceled", "digest", hash)
+			s.log.Infow("msg", "download canceled", "digest", wantChecksum)
 			return
 		}
 
 		http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Verify the checksum
+	if got, want := fmt.Sprintf("%x", 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)
+		return
+	}
+
+	// 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", info.FileName))
+	w.Header().Set("Content-Length", strconv.FormatInt(info.Size, 10))
+
+	if _, err := io.Copy(w, buf); err != nil {
+		http.Error(w, sl.LogAndMaskErr(err, s.log).Error(), http.StatusInternalServerError)
+		return
 	}
 
-	s.log.Infow("msg", "download finished", "digest", hash, "size", bytefmt.ByteSize(uint64(info.Size)))
+	s.log.Infow("msg", "download finished", "digest", wantChecksum, "size", bytefmt.ByteSize(uint64(info.Size)))
 }

app/artifact-cas/internal/service/service.go

@@ -40,7 +40,7 @@ func (s *commonService) loadBackend(ctx context.Context, providerType, secretID
 	// get the OCI provider from the map
 	p, ok := s.backends[providerType]
 	if !ok || p == nil {
-		return nil, kerrors.NotFound("backend provider", providerType)
+		return nil, kerrors.NotFound("backend provider", fmt.Sprintf("backend %q not found", providerType))
 	}
 
 	s.log.Infow("msg", "selected provider", "provider", providerType)

app/cli/cmd/casbackend.go

@@ -41,7 +41,7 @@ func newCASBackendAddCmd() *cobra.Command {
 	cmd.PersistentFlags().Bool("default", false, "set the backend as default in your organization")
 	cmd.PersistentFlags().String("description", "", "descriptive information for this registration")
 
-	cmd.AddCommand(newCASBackendAddOCICmd())
+	cmd.AddCommand(newCASBackendAddOCICmd(), newCASBackendAddAzureBlobStorageCmd())
 	return cmd
 }
 
@@ -54,7 +54,7 @@ func newCASBackendUpdateCmd() *cobra.Command {
 	cmd.PersistentFlags().Bool("default", false, "set the backend as default in your organization")
 	cmd.PersistentFlags().String("description", "", "descriptive information for this registration")
 
-	cmd.AddCommand(newCASBackendUpdateOCICmd(), newCASBackendUpdateInlineCmd())
+	cmd.AddCommand(newCASBackendUpdateOCICmd(), newCASBackendUpdateInlineCmd(), newCASBackendUpdateAzureBlobCmd())
 	return cmd
 }
 

app/cli/cmd/casbackend_add_azureblob.go

@@ -0,0 +1,91 @@
+//
+// Copyright 2023 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 cmd
+
+import (
+	"fmt"
+
+	"github.com/chainloop-dev/chainloop/app/cli/internal/action"
+	"github.com/chainloop-dev/chainloop/internal/blobmanager/azureblob"
+	"github.com/go-kratos/kratos/v2/log"
+	"github.com/spf13/cobra"
+)
+
+func newCASBackendAddAzureBlobStorageCmd() *cobra.Command {
+	var storageAccountName, tenantID, clientID, clientSecret, container string
+	cmd := &cobra.Command{
+		Use:   "azure-blob",
+		Short: "Register a Azure Blob Storage CAS Backend",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			// If we are setting the default, we list existing CAS backends
+			// and ask the user to confirm the rewrite
+			isDefault, err := cmd.Flags().GetBool("default")
+			cobra.CheckErr(err)
+
+			description, err := cmd.Flags().GetString("description")
+			cobra.CheckErr(err)
+
+			if isDefault {
+				if confirmed, err := confirmDefaultCASBackendOverride(actionOpts, ""); err != nil {
+					return err
+				} else if !confirmed {
+					log.Info("Aborting...")
+					return nil
+				}
+			}
+
+			opts := &action.NewCASBackendAddOpts{
+				Location:    fmt.Sprintf("%s/%s", storageAccountName, container),
+				Provider:    azureblob.ProviderID,
+				Description: description,
+				Credentials: map[string]any{
+					"tenantID":     tenantID,
+					"clientID":     clientID,
+					"clientSecret": clientSecret,
+				},
+				Default: isDefault,
+			}
+
+			res, err := action.NewCASBackendAdd(actionOpts).Run(opts)
+			if err != nil {
+				return err
+			} else if res == nil {
+				return nil
+			}
+
+			return encodeOutput([]*action.CASBackendItem{res}, casBackendListTableOutput)
+		},
+	}
+
+	cmd.Flags().StringVar(&storageAccountName, "storage-account", "", "Storage Account Name")
+	err := cmd.MarkFlagRequired("storage-account")
+	cobra.CheckErr(err)
+
+	cmd.Flags().StringVar(&tenantID, "tenant", "", "Active Directory Tenant ID")
+	err = cmd.MarkFlagRequired("tenant")
+	cobra.CheckErr(err)
+
+	cmd.Flags().StringVar(&clientID, "client-id", "", "Service Principal Client ID")
+	err = cmd.MarkFlagRequired("client-id")
+	cobra.CheckErr(err)
+
+	cmd.Flags().StringVar(&clientSecret, "client-secret", "", "Service Principal Client Secret")
+	err = cmd.MarkFlagRequired("client-secret")
+	cobra.CheckErr(err)
+
+	cmd.Flags().StringVar(&container, "container", "chainloop", "Storage Container Name")
+	return cmd
+}