feat(blobmanager): add managed CAS backend via S3 Access Points (#3121)

Signed-off-by: Jose I. Paris <jiparis@chainloop.dev>

Author: jiparis

.gitattributes

@@ -4,4 +4,7 @@ app/controlplane/pkg/data/ent/migrate/** linguist-generated=false
 app/controlplane/pkg/data/ent/migrate/** linguist-detectable=true
 app/controlplane/pkg/data/ent/schema/* linguist-generated=false
 app/controlplane/pkg/data/ent/schema/* linguist-detectable=true
-app/controlplane/api/gen/jsonschema/** linguist-generated=true
+app/controlplane/api/gen/jsonschema/** linguist-generated=true
+*.pb.go linguist-generated=true
+*.pb.validate.go linguist-generated=true
+wire_gen.go linguist-generated=true

app/artifact-cas/internal/server/grpc_test.go

@@ -97,7 +97,7 @@ func TestJWTAuthFunc(t *testing.T) {
 
 			b, err := robotaccount.NewBuilder(opts...)
 			require.NoError(t, err)
-			token, err := b.GenerateJWT("backend-type", "secret-id", tc.audience, robotaccount.Downloader, 0)
+			token, err := b.GenerateJWT("backend-type", "secret-id", tc.audience, robotaccount.Downloader, 0, "org-id")
 			require.NoError(t, err)
 
 			// add bearer token to context

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

@@ -64,7 +64,7 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 	defer span.End()
 
 	// Get auth info and check that it's an uploader token
-	info, err := infoFromAuth(ctx)
+	info, err := casJWT.InfoFromAuth(ctx)
 	if err != nil {
 		return err
 	}
@@ -145,7 +145,7 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.
 	ctx := stream.Context()
 	ctx, span := otelx.Start(ctx, byteStreamTracer, "ByteStreamService.Read")
 	defer span.End()
-	info, err := infoFromAuth(ctx)
+	info, err := casJWT.InfoFromAuth(ctx)
 	if err != nil {
 		return err
 	}

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

@@ -52,7 +52,7 @@ func (s *DownloadService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 	ctx, span := otelx.Start(ctx, downloadTracer, "DownloadService.ServeHTTP")
 	defer span.End()
-	auth, err := infoFromAuth(ctx)
+	auth, err := casJWT.InfoFromAuth(ctx)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusUnauthorized)
 		return

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

@@ -19,6 +19,7 @@ import (
 	"context"
 
 	v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
+	casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas"
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/chainloop-dev/chainloop/pkg/otelx"
 	sl "github.com/chainloop-dev/chainloop/pkg/servicelogger"
@@ -43,7 +44,7 @@ func (s *ResourceService) Describe(ctx context.Context, req *v1.ResourceServiceD
 	ctx, span := otelx.Start(ctx, resourceTracer, "ResourceService.Describe")
 	defer span.End()
 
-	info, err := infoFromAuth(ctx)
+	info, err := casJWT.InfoFromAuth(ctx)
 	if err != nil {
 		return nil, err
 	}

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

@@ -21,12 +21,10 @@ import (
 	"fmt"
 	"syscall"
 
-	casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas"
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/chainloop-dev/chainloop/pkg/servicelogger"
 	kerrors "github.com/go-kratos/kratos/v2/errors"
 	"github.com/go-kratos/kratos/v2/log"
-	"github.com/go-kratos/kratos/v2/middleware/auth/jwt"
 	"github.com/google/wire"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -104,30 +102,3 @@ func isClientDisconnect(err error) bool {
 
 	return false
 }
-
-// Extract the JWT claims from the context, note that the JWT verification has happened in the middleware
-func infoFromAuth(ctx context.Context) (*casJWT.Claims, error) {
-	rawClaims, ok := jwt.FromContext(ctx)
-	if !ok {
-		return nil, kerrors.Unauthorized("cas", "missing authentication information")
-	}
-
-	claims, ok := rawClaims.(*casJWT.Claims)
-	if !ok {
-		return nil, kerrors.Unauthorized("cas", "invalid authentication information")
-	}
-
-	if claims.StoredSecretID == "" {
-		return nil, kerrors.Unauthorized("cas", "missing secret reference")
-	}
-
-	if claims.BackendType == "" {
-		return nil, kerrors.Unauthorized("cas", "missing backend type")
-	}
-
-	if claims.Role != casJWT.Uploader && claims.Role != casJWT.Downloader {
-		return nil, kerrors.Unauthorized("cas", "invalid role")
-	}
-
-	return claims, nil
-}

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

@@ -24,90 +24,15 @@ import (
 	"syscall"
 	"testing"
 
-	casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas"
 	backend "github.com/chainloop-dev/chainloop/pkg/blobmanager"
 	"github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks"
 	kerrors "github.com/go-kratos/kratos/v2/errors"
-	jwtm "github.com/go-kratos/kratos/v2/middleware/auth/jwt"
-	"github.com/golang-jwt/jwt/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
-func TestInfoFromAuth(t *testing.T) {
-	testCases := []struct {
-		name string
-		// input
-		claims  jwt.Claims
-		wantErr bool
-	}{
-		{
-			name: "valid claims downloader",
-			claims: &casJWT.Claims{
-				Role:           casJWT.Downloader,
-				StoredSecretID: "test",
-				BackendType:    "backend-type",
-			},
-		},
-		{
-			name: "valid claims uploader",
-			claims: &casJWT.Claims{
-				Role:           casJWT.Uploader,
-				StoredSecretID: "test",
-				BackendType:    "backend-type",
-			},
-		},
-		{
-			name: "invalid role",
-			claims: &casJWT.Claims{
-				Role:           "invalid",
-				StoredSecretID: "test",
-				BackendType:    "backend-type",
-			},
-			wantErr: true,
-		},
-		{
-			name: "missing secretID",
-			claims: &casJWT.Claims{
-				Role:        "test",
-				BackendType: "backend-type",
-			},
-			wantErr: true,
-		},
-		{
-			name: "missing role",
-			claims: &casJWT.Claims{
-				StoredSecretID: "test",
-				BackendType:    "backend-type",
-			},
-			wantErr: true,
-		},
-		{
-			name: "missing backend type",
-			claims: &casJWT.Claims{
-				StoredSecretID: "test",
-				Role:           "test",
-			},
-			wantErr: true,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			info, err := infoFromAuth(jwtm.NewContext(context.Background(), tc.claims))
-			if tc.wantErr {
-				assert.Error(t, err)
-				return
-			}
-
-			assert.NoError(t, err)
-			assert.Equal(t, tc.claims, info)
-		})
-	}
-}
-
 func TestLoadBackend(t *testing.T) {
 	testCases := []struct {
 		name string

app/controlplane/internal/dispatcher/dispatcher.go

@@ -92,7 +92,7 @@ func (d *FanOutDispatcher) Run(ctx context.Context, opts *RunOpts) error {
 	}
 
 	// 2. Hydrate the dispatch queue with the actual inputs
-	if err := d.loadInputs(ctx, queue, opts.Envelope, opts.DownloadBackendType, opts.DownloadSecretName); err != nil {
+	if err := d.loadInputs(ctx, queue, opts.Envelope, opts.DownloadBackendType, opts.DownloadSecretName, opts.OrgID); err != nil {
 		return fmt.Errorf("loading materials: %w", err)
 	}
 
@@ -198,7 +198,7 @@ func (d *FanOutDispatcher) initDispatchQueue(ctx context.Context, orgID, workflo
 }
 
 // Load the inputs for the dispatchItem, both materials and attestation
-func (d *FanOutDispatcher) loadInputs(ctx context.Context, queue dispatchQueue, att *dsse.Envelope, backendType, secretName string) error {
+func (d *FanOutDispatcher) loadInputs(ctx context.Context, queue dispatchQueue, att *dsse.Envelope, backendType, secretName, orgID string) error {
 	if att == nil {
 		return fmt.Errorf("attestation is nil")
 	}
@@ -252,8 +252,12 @@ func (d *FanOutDispatcher) loadInputs(ctx context.Context, queue dispatchQueue,
 			if item.plugin.IsSubscribedTo(material.Type) {
 				// It's a downloadable and has not been downloaded yet
 				if !downloaded && material.Hash != nil && material.UploadedToCAS {
+					orgUUID, err := uuid.Parse(orgID)
+					if err != nil {
+						return fmt.Errorf("parsing org id: %w", err)
+					}
 					buf := bytes.NewBuffer(nil)
-					if err := d.casClient.Download(ctx, backendType, secretName, buf, material.Hash.String()); err != nil {
+					if err := d.casClient.Download(ctx, backendType, secretName, orgUUID, buf, material.Hash.String()); err != nil {
 						return fmt.Errorf("downloading from CAS: %w", err)
 					}
 

app/controlplane/internal/dispatcher/dispatcher_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1"
 	mockedSDK "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1/mocks"
 	"github.com/go-kratos/kratos/v2/log"
+	"github.com/google/uuid"
 	"github.com/secure-systems-lab/go-securesystemslib/dsse"
 
 	"github.com/chainloop-dev/chainloop/app/controlplane/pkg/biz/testhelpers"
@@ -57,7 +58,7 @@ func (s *dispatcherTestSuite) TestLoadInputsEnvelope() {
 	s.ociIntegrationBackend.(*mockedSDK.FanOut).On("IsSubscribedTo", "SBOM_CYCLONEDX_JSON").Return(false)
 	s.ociIntegrationBackend.(*mockedSDK.FanOut).On("String").Return("mocked-integration")
 
-	err = s.dispatcher.loadInputs(context.TODO(), queue, envelope, "backend-type", "secret-name")
+	err = s.dispatcher.loadInputs(context.TODO(), queue, envelope, "backend-type", "secret-name", uuid.NewString())
 	assert.NoError(s.T(), err)
 
 	// Only one integration is registered
@@ -101,14 +102,14 @@ func (s *dispatcherTestSuite) TestLoadInputsMaterials() {
 	require.NoError(s.T(), err)
 
 	// Simulate SBOM download
-	s.casClient.On("Download", mock.Anything, "backend-type", "secret-name", mock.Anything, mock.Anything).
+	s.casClient.On("Download", mock.Anything, "backend-type", "secret-name", mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Run(func(args mock.Arguments) {
 		buf := bytes.NewBuffer([]byte("SBOM Content"))
-		_, err := io.Copy(args.Get(3).(io.Writer), buf)
+		_, err := io.Copy(args.Get(4).(io.Writer), buf)
 		s.NoError(err)
 	})
 
-	err = s.dispatcher.loadInputs(context.TODO(), queue, envelope, "backend-type", "secret-name")
+	err = s.dispatcher.loadInputs(context.TODO(), queue, envelope, "backend-type", "secret-name", uuid.NewString())
 	assert.NoError(s.T(), err)
 	require.Len(s.T(), queue, 3)
 

app/controlplane/internal/service/attestation.go

@@ -494,7 +494,7 @@ func (s *AttestationService) GetUploadCreds(ctx context.Context, req *cpAPI.Atte
 	// Return the backend information and associated credentials (if applicable)
 	resp := &cpAPI.AttestationServiceGetUploadCredsResponse_Result{Backend: bizCASBackendToPb(backend)}
 	if backend.SecretName != "" {
-		ref := &biz.CASCredsOpts{BackendType: string(backend.Provider), SecretPath: backend.SecretName, Role: casJWT.Uploader, MaxBytes: backend.Limits.MaxBytes}
+		ref := &biz.CASCredsOpts{BackendType: string(backend.Provider), SecretPath: backend.SecretName, Role: casJWT.Uploader, MaxBytes: backend.Limits.MaxBytes, OrgID: backend.OrganizationID}
 		t, err := s.casCredsUseCase.GenerateTemporaryCredentials(ref)
 		if err != nil {
 			return nil, handleUseCaseErr(err, s.log)