Merge pull request #25 from mstg/keykeeper-fixes-1

peridot/keykeeper/v1/keywarming.go

@@ -38,8 +38,6 @@ import (
 	"fmt"
 	"github.com/ProtonMail/gopenpgp/v2/crypto"
 	"github.com/google/uuid"
-	"io/ioutil"
-	"os"
 	"os/exec"
 	"peridot.resf.org/peridot/db/models"
 	"peridot.resf.org/utils"
@@ -67,14 +65,6 @@ func gpgCmdEnv(cmd *exec.Cmd) *exec.Cmd {
 	return cmd
 }
 
-func (s *Server) deleteGpgKey(keyId string) error {
-	out, err := logCmdRun(gpgCmdEnv(exec.Command("gpg", "--batch", "--yes", "--delete-secret-and-public-key", keyId)))
-	if err != nil {
-		s.log.Errorf("failed to delete gpg key: %s", out.String())
-	}
-	return err
-}
-
 func (s *Server) importGpgKey(armoredKey string) error {
 	cmd := gpgCmdEnv(exec.Command("gpg", "--batch", "--yes", "--import", "-"))
 	cmd.Stdin = strings.NewReader(armoredKey)
@@ -85,41 +75,13 @@ func (s *Server) importGpgKey(armoredKey string) error {
 	return err
 }
 
-func (s *Server) importRpmKey(publicKey string) error {
-	tmpFile, err := ioutil.TempFile("/tmp", "peridot-key-")
-	if err != nil {
-		return err
-	}
-	defer os.Remove(tmpFile.Name())
-	_, err = tmpFile.Write([]byte(publicKey))
-	if err != nil {
-		return err
-	}
-	cmd := gpgCmdEnv(exec.Command("rpm", "--import", tmpFile.Name()))
-	out, err := logCmdRun(cmd)
-	if err != nil {
-		s.log.Errorf("failed to import rpm key: %s", out.String())
-	}
-	return err
-}
-
 // WarmGPGKey warms up a specific GPG key
 // This involves shelling out to GPG to import the key
 func (s *Server) WarmGPGKey(key string, armoredKey string, gpgKey *crypto.Key, db *models.Key) (*LoadedKey, error) {
-	cachedKey := s.keys[key]
+	cachedKeyAny, ok := s.keys.Load(key)
 	// This means that the key is already loaded
-	// We need to delete and replace it
+	if ok {
-	if cachedKey != nil {
+		return cachedKeyAny.(*LoadedKey), nil
-		cachedKey.Lock()
-		defer cachedKey.Unlock()
-
-		keyId := gpgKey.GetHexKeyID()
-		err := s.deleteGpgKey(keyId)
-		if err != nil {
-			return nil, err
-		}
-
-		cachedKey.gpgId = keyId
 	}
 
 	err := s.importGpgKey(armoredKey)
@@ -127,26 +89,20 @@ func (s *Server) WarmGPGKey(key string, armoredKey string, gpgKey *crypto.Key, d
 		return nil, err
 	}
 
-	err = s.importRpmKey(db.PublicKey)
+	cachedKey := &LoadedKey{
-	if err != nil {
+		keyUuid: db.ID,
-		return nil, err
+		gpgId:   gpgKey.GetHexKeyID(),
 	}
+	s.keys.Store(key, cachedKey)
 
-	if cachedKey == nil {
+	return cachedKey, nil
-		s.keys[key] = &LoadedKey{
-			keyUuid: db.ID,
-			gpgId:   gpgKey.GetHexKeyID(),
-		}
-	}
-
-	return s.keys[key], nil
 }
 
 // EnsureGPGKey ensures that the key is loaded
 func (s *Server) EnsureGPGKey(key string) (*LoadedKey, error) {
-	cachedKey := s.keys[key]
+	cachedKeyAny, ok := s.keys.Load(key)
-	if cachedKey != nil {
+	if ok {
-		return cachedKey, nil
+		return cachedKeyAny.(*LoadedKey), nil
 	}
 
 	// Key not found in cache, fetch from database

peridot/keykeeper/v1/server.go

@@ -66,7 +66,7 @@ type Server struct {
 	worker       worker.Worker
 	temporal     client.Client
 	stores       map[string]store.Store
-	keys         map[string]*LoadedKey
+	keys         *sync.Map
 	defaultStore string
 }
 
@@ -82,13 +82,15 @@ func NewServer(db peridotdb.Access, c client.Client) (*Server, error) {
 	}
 
 	return &Server{
-		log:          logrus.New(),
+		log:     logrus.New(),
-		db:           db,
+		db:      db,
-		storage:      storage,
+		storage: storage,
-		worker:       worker.New(c, TaskQueue, worker.Options{}),
+		worker: worker.New(c, TaskQueue, worker.Options{
+			DeadlockDetectionTimeout: 15 * time.Minute,
+		}),
 		temporal:     c,
 		stores:       map[string]store.Store{"awssm": sm},
-		keys:         map[string]*LoadedKey{},
+		keys:         &sync.Map{},
 		defaultStore: "awssm",
 	}, nil
 }

peridot/keykeeper/v1/sign.go

@@ -114,7 +114,7 @@ func (s *Server) SignArtifactsWorkflow(ctx workflow.Context, artifacts models.Ta
 		signArtifactCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 			ScheduleToStartTimeout: 10 * time.Hour,
 			StartToCloseTimeout:    24 * time.Hour,
-			HeartbeatTimeout:       time.Minute,
+			HeartbeatTimeout:       10 * time.Minute,
 			TaskQueue:              TaskQueue,
 		})
 		futures = append(futures, peridotworkflow.FutureContext{
@@ -188,90 +188,60 @@ func (s *Server) SignArtifactActivity(ctx context.Context, artifactId string, ke
 
 	switch ext {
 	case ".rpm":
-		rpmSign := func() (*keykeeperpb.SignedArtifact, error) {
+		var outBuf bytes.Buffer
-			var outBuf bytes.Buffer
+		opts := []string{
-			opts := []string{
+			"--define", "_gpg_name " + keyName,
-				"--define", "_gpg_name " + keyName,
+			"--define", "_peridot_keykeeper_key " + key.keyUuid.String(),
-				"--define", "_peridot_keykeeper_key " + key.keyUuid.String(),
+			"--addsign", localPath,
-				"--addsign", localPath,
-			}
-			cmd := gpgCmdEnv(exec.Command("rpm", opts...))
-			cmd.Stdout = &outBuf
-			cmd.Stderr = &outBuf
-			err := cmd.Run()
-			if err != nil {
-				s.log.Errorf("failed to sign artifact %s: %v", artifact.Name, err)
-				statusErr := status.New(codes.Internal, "failed to sign artifact")
-				statusErr, err2 := statusErr.WithDetails(&errdetails.ErrorInfo{
-					Reason: "rpmsign-failed",
-					Domain: "keykeeper.peridot.resf.org",
-					Metadata: map[string]string{
-						"logs": outBuf.String(),
-						"err":  err.Error(),
-					},
-				})
-				if err2 != nil {
-					s.log.Errorf("failed to add error details to status: %v", err2)
-				}
-				return nil, statusErr.Err()
-			}
-			_, err = s.storage.PutObject(newObjectKey, localPath)
-			if err != nil {
-				s.log.Errorf("failed to upload artifact %s: %v", newObjectKey, err)
-				return nil, fmt.Errorf("failed to upload artifact %s: %v", newObjectKey, err)
-			}
-
-			f, err := os.Open(localPath)
-			if err != nil {
-				return nil, err
-			}
-
-			hasher := sha256.New()
-			_, err = io.Copy(hasher, f)
-			if err != nil {
-				return nil, err
-			}
-			hash := hex.EncodeToString(hasher.Sum(nil))
-
-			err = s.db.CreateTaskArtifactSignature(artifact.ID.String(), key.keyUuid.String(), hash)
-			if err != nil {
-				s.log.Errorf("failed to create task artifact signature: %v", err)
-				return nil, fmt.Errorf("failed to create task artifact signature: %v", err)
-			}
-
-			return &keykeeperpb.SignedArtifact{
-				Path:       newObjectKey,
-				HashSha256: hash,
-			}, nil
 		}
-		verifySig := func() error {
+		cmd := gpgCmdEnv(exec.Command("rpm", opts...))
-			opts := []string{
+		cmd.Stdout = &outBuf
-				"--define", "_gpg_name " + keyName,
+		cmd.Stderr = &outBuf
-				"--define", "_peridot_keykeeper_key " + key.keyUuid.String(),
+		err := cmd.Run()
-				"--checksig", localPath,
+		if err != nil {
+			s.log.Errorf("failed to sign artifact %s: %v", artifact.Name, err)
+			statusErr := status.New(codes.Internal, "failed to sign artifact")
+			statusErr, err2 := statusErr.WithDetails(&errdetails.ErrorInfo{
+				Reason: "rpmsign-failed",
+				Domain: "keykeeper.peridot.resf.org",
+				Metadata: map[string]string{
+					"logs": outBuf.String(),
+					"err":  err.Error(),
+				},
+			})
+			if err2 != nil {
+				s.log.Errorf("failed to add error details to status: %v", err2)
 			}
-			cmd := gpgCmdEnv(exec.Command("rpm", opts...))
+			return nil, statusErr.Err()
-			cmd.Stdout = os.Stdout
-			cmd.Stderr = os.Stderr
-			err := cmd.Run()
-			if err != nil {
-				s.log.Errorf("failed to verify artifact %s: %v", artifact.Name, err)
-				return fmt.Errorf("failed to verify artifact %s: %v", artifact.Name, err)
-			}
-			return nil
 		}
-		var tries int
+		_, err = s.storage.PutObject(newObjectKey, localPath)
-		for {
+		if err != nil {
-			res, _ := rpmSign()
+			s.log.Errorf("failed to upload artifact %s: %v", newObjectKey, err)
-			err := verifySig()
+			return nil, fmt.Errorf("failed to upload artifact %s: %v", newObjectKey, err)
-			if err == nil {
-				return res, nil
-			}
-			if err != nil && tries > 3 {
-				return nil, err
-			}
-			tries++
 		}
+
+		f, err := os.Open(localPath)
+		if err != nil {
+			return nil, err
+		}
+
+		hasher := sha256.New()
+		_, err = io.Copy(hasher, f)
+		if err != nil {
+			return nil, err
+		}
+		hash := hex.EncodeToString(hasher.Sum(nil))
+
+		err = s.db.CreateTaskArtifactSignature(artifact.ID.String(), key.keyUuid.String(), hash)
+		if err != nil {
+			s.log.Errorf("failed to create task artifact signature: %v", err)
+			return nil, fmt.Errorf("failed to create task artifact signature: %v", err)
+		}
+
+		return &keykeeperpb.SignedArtifact{
+			Path:       newObjectKey,
+			HashSha256: hash,
+		}, nil
 	default:
 		s.log.Infof("skipping artifact %s, extension %s not supported", artifact.Name, ext)
 		return nil, ErrUnsupportedExtension

peridot/lookaside/s3/s3.go

@@ -41,16 +41,16 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/go-git/go-billy/v5"
 	"github.com/spf13/viper"
-	"io"
 	"io/ioutil"
 	"os"
 	"peridot.resf.org/peridot/lookaside"
 )
 
 type Storage struct {
-	bucket   string
+	bucket     string
-	uploader *s3manager.Uploader
+	uploader   *s3manager.Uploader
-	fs       billy.Filesystem
+	downloader *s3manager.Downloader
+	fs         billy.Filesystem
 }
 
 func New(fs billy.Filesystem) (*Storage, error) {
@@ -81,34 +81,32 @@ func New(fs billy.Filesystem) (*Storage, error) {
 		return nil, err
 	}
 	uploader := s3manager.NewUploader(sess)
+	downloader := s3manager.NewDownloader(sess)
 
 	return &Storage{
-		bucket:   viper.GetString("s3-bucket"),
+		bucket:     viper.GetString("s3-bucket"),
-		uploader: uploader,
+		uploader:   uploader,
-		fs:       fs,
+		downloader: downloader,
+		fs:         fs,
 	}, nil
 }
 
 func (s *Storage) DownloadObject(objectName string, path string) error {
-	obj, err := s.uploader.S3.GetObject(&s3.GetObjectInput{
+	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
-		Bucket: aws.String(s.bucket),
-		Key:    aws.String(objectName),
-	})
 	if err != nil {
 		return err
 	}
+	defer f.Close()
 
-	f, err := s.fs.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
+	_, err = s.downloader.Download(
-	if err != nil {
+		f,
-		return err
+		&s3.GetObjectInput{
-	}
+			Bucket: aws.String(s.bucket),
+			Key:    aws.String(objectName),
+		},
+	)
 
-	_, err = io.Copy(f, obj.Body)
+	return err
-	if err != nil {
-		return err
-	}
-
-	return nil
 }
 
 func (s *Storage) ReadObject(objectName string) ([]byte, error) {