refactor(CAS proxy): multiple CAS backend providers support (#359)

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

Author: migmartri

.golangci.yml

@@ -71,4 +71,8 @@ severity:
     - linters:
         - staticcheck
       text: "SA1019:"
-      severity: info
+      severity: info
+issues:
+  exclude-rules:
+    - path: _test\.go
+      text: "Potential hardcoded credentials"

app/artifact-cas/api/cas/v1/status.pb.go

@@ -17,34 +17,31 @@ syntax = "proto3";
 
 package cas.v1;
 
-option go_package = "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1;v1";
-
 import "google/api/annotations.proto";
 
+option go_package = "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1;v1";
+
 service StatusService {
-	rpc Infoz (InfozRequest) returns (InfozResponse) {
-		option (google.api.http) = {
-            get: "/infoz"
-        };
-	}
-	rpc Statusz (StatuszRequest) returns (StatuszResponse) {
-		option (google.api.http) = {
-            get: "/statusz"
-        };
-	}
+  rpc Infoz(InfozRequest) returns (InfozResponse) {
+    option (google.api.http) = {get: "/infoz"};
+  }
+  rpc Statusz(StatuszRequest) returns (StatuszResponse) {
+    option (google.api.http) = {get: "/statusz"};
+  }
 }
 
-message InfozRequest { }
+message InfozRequest {}
 
 message InfozResponse {
-	string version = 1;
+  string version = 1;
+  repeated string backends = 2;
 }
 
 message StatuszRequest {
-	// Parameter that can be used by readiness probes
-	// The main difference is that readiness probes will take into account that all
-	// dependent services are up and ready
-	bool readiness = 1;
+  // Parameter that can be used by readiness probes
+  // The main difference is that readiness probes will take into account that all
+  // dependent services are up and ready
+  bool readiness = 1;
 }
 
 message StatuszResponse {}

app/artifact-cas/api/cas/v1/status.proto

@@ -24,6 +24,8 @@ 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/oci"
 	"github.com/chainloop-dev/chainloop/internal/credentials"
 	"github.com/chainloop-dev/chainloop/internal/credentials/manager"
 	"github.com/chainloop-dev/chainloop/internal/servicelogger"
@@ -56,19 +58,36 @@ func init() {
 	flag.StringVar(&flagconf, "conf", "../../configs", "config path, eg: -conf config.yaml")
 }
 
-func newApp(logger log.Logger, gs *grpc.Server, hs *http.Server, ms *server.HTTPMetricsServer) *kratos.App {
-	return kratos.New(
-		kratos.ID(id),
-		kratos.Name(Name),
-		kratos.Version(Version),
-		kratos.Metadata(map[string]string{}),
-		kratos.Logger(logger),
-		kratos.Server(
-			gs,
-			hs,
-			ms,
+type app struct {
+	*kratos.App
+	backend.Providers
+}
+
+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)
+	return backend.Providers{
+		p.ID(): p,
+	}
+}
+
+func newApp(logger log.Logger, gs *grpc.Server, hs *http.Server, ms *server.HTTPMetricsServer, providers backend.Providers) *app {
+	return &app{
+		kratos.New(
+			kratos.ID(id),
+			kratos.Name(Name),
+			kratos.Version(Version),
+			kratos.Metadata(map[string]string{}),
+			kratos.Logger(logger),
+			kratos.Server(
+				gs,
+				hs,
+				ms,
+			),
 		),
-	)
+		providers,
+	}
 }
 
 func main() {
@@ -115,6 +134,10 @@ func main() {
 	}
 	defer cleanup()
 
+	for k := range app.Providers {
+		_ = logger.Log(log.LevelInfo, "msg", "CAS backend provider loaded", "provider", k)
+	}
+
 	// start and wait for stop signal
 	if err := app.Run(); err != nil {
 		panic(err)

app/artifact-cas/cmd/main.go

@@ -24,22 +24,18 @@ 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/internal/blobmanager"
-	"github.com/chainloop-dev/chainloop/internal/blobmanager/oci"
 	"github.com/chainloop-dev/chainloop/internal/credentials"
-	"github.com/go-kratos/kratos/v2"
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/google/wire"
 )
 
 // wireApp init kratos application.
-func wireApp(*conf.Server, *conf.Auth, credentials.Reader, log.Logger) (*kratos.App, func(), error) {
+func wireApp(*conf.Server, *conf.Auth, credentials.Reader, log.Logger) (*app, func(), error) {
 	panic(
 		wire.Build(
 			server.ProviderSet,
 			service.ProviderSet,
-			wire.Bind(new(backend.Provider), new(*oci.BackendProvider)),
-			oci.NewBackendProvider,
+			loadCASBackendProviders,
 			newApp,
 			serviceOpts,
 		),

app/artifact-cas/cmd/wire.go

@@ -25,6 +25,7 @@ import (
 	v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf"
 	"github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service"
+	backend "github.com/chainloop-dev/chainloop/internal/blobmanager"
 	casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas"
 	"github.com/getsentry/sentry-go"
 	"github.com/go-kratos/kratos/v2/errors"
@@ -43,7 +44,7 @@ import (
 )
 
 // NewGRPCServer new a gRPC server.
-func NewGRPCServer(c *conf.Server, authConf *conf.Auth, byteService *service.ByteStreamService, rSvc *service.ResourceService, logger log.Logger) (*grpc.Server, error) {
+func NewGRPCServer(c *conf.Server, authConf *conf.Auth, byteService *service.ByteStreamService, rSvc *service.ResourceService, providers backend.Providers, logger log.Logger) (*grpc.Server, error) {
 	log := log.NewHelper(logger)
 	// Load the key on initialization instead of on every request
 	// TODO: implement jwks endpoint
@@ -119,7 +120,7 @@ func NewGRPCServer(c *conf.Server, authConf *conf.Auth, byteService *service.Byt
 
 	bytestream.RegisterByteStreamServer(srv.Server, byteService)
 	v1.RegisterResourceServiceServer(srv.Server, rSvc)
-	v1.RegisterStatusServiceServer(srv.Server, service.NewStatusService(Version))
+	v1.RegisterStatusServiceServer(srv.Server, service.NewStatusService(Version, providers))
 
 	// Register and set metrics to 0
 	grpc_prometheus.Register(srv.Server)

app/artifact-cas/cmd/wire_gen.go

@@ -96,7 +96,7 @@ func TestJWTAuthFunc(t *testing.T) {
 
 			b, err := robotaccount.NewBuilder(opts...)
 			require.NoError(t, err)
-			token, err := b.GenerateJWT("secret-id", tc.audience, robotaccount.Downloader)
+			token, err := b.GenerateJWT("backend-type", "secret-id", tc.audience, robotaccount.Downloader)
 			require.NoError(t, err)
 
 			// add bearer token to context
@@ -127,6 +127,7 @@ func TestJWTAuthFunc(t *testing.T) {
 				claims := infoFromAuth(ctx, t)
 				assert.NoError(t, claims.Valid())
 				assert.Equal(t, "secret-id", claims.StoredSecretID)
+				assert.Equal(t, "backend-type", claims.BackendType)
 				assert.Equal(t, robotaccount.Downloader, claims.Role)
 				assert.Equal(t, "my-issuer", claims.Issuer)
 				assert.Contains(t, claims.Audience, "artifact-cas.chainloop")

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

@@ -27,6 +27,7 @@ import (
 	nhttp "net/http"
 
 	api "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1"
+	backend "github.com/chainloop-dev/chainloop/internal/blobmanager"
 	"github.com/go-kratos/kratos/v2/log"
 	jwtMiddleware "github.com/go-kratos/kratos/v2/middleware/auth/jwt"
 	"github.com/go-kratos/kratos/v2/middleware/logging"
@@ -35,7 +36,7 @@ import (
 )
 
 // NewHTTPServer new a HTTP server.
-func NewHTTPServer(c *conf.Server, authConf *conf.Auth, downloadSvc *service.DownloadService, logger log.Logger) (*http.Server, error) {
+func NewHTTPServer(c *conf.Server, authConf *conf.Auth, downloadSvc *service.DownloadService, providers backend.Providers, logger log.Logger) (*http.Server, error) {
 	var opts = []http.ServerOption{
 		http.Middleware(
 			recovery.Recovery(),
@@ -68,7 +69,7 @@ func NewHTTPServer(c *conf.Server, authConf *conf.Auth, downloadSvc *service.Dow
 	srv := http.NewServer(opts...)
 
 	srv.Handle(service.DownloadPath, authFromQueryMiddleware(loadPublicKey(rawKey), casJWT.SigningMethod, downloadSvc))
-	api.RegisterStatusServiceHTTPServer(srv, service.NewStatusService(Version))
+	api.RegisterStatusServiceHTTPServer(srv, service.NewStatusService(Version, providers))
 	return srv, nil
 }
 

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

@@ -41,7 +41,7 @@ type ByteStreamService struct {
 	*commonService
 }
 
-func NewByteStreamService(bp backend.Provider, opts ...NewOpt) *ByteStreamService {
+func NewByteStreamService(bp backend.Providers, opts ...NewOpt) *ByteStreamService {
 	return &ByteStreamService{
 		commonService: newCommonService(bp, opts...),
 	}
@@ -70,9 +70,10 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro
 		return kerrors.BadRequest("resource name", err.Error())
 	}
 
-	// Load OCI backend based on a reference stored in the token
-	backend, err := s.backendP.FromCredentials(ctx, info.StoredSecretID)
-	if err != nil {
+	backend, err := s.loadBackend(ctx, info.BackendType, info.StoredSecretID)
+	if err != nil && kerrors.IsNotFound(err) {
+		return err
+	} else if err != nil {
 		return sl.LogAndMaskErr(err, s.log)
 	}
 
@@ -142,9 +143,10 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream.
 		return kerrors.BadRequest("resource name", "empty resource name")
 	}
 
-	// Retrieve the OCI backend from where to download the file
-	backend, err := s.backendP.FromCredentials(ctx, info.StoredSecretID)
-	if err != nil {
+	backend, err := s.loadBackend(ctx, info.BackendType, info.StoredSecretID)
+	if err != nil && kerrors.IsNotFound(err) {
+		return err
+	} else if err != nil {
 		return sl.LogAndMaskErr(err, s.log)
 	}