image: add image signature importer controller

Author: mfojtik

pkg/cmd/server/origin/controller/config.go

@@ -48,8 +48,9 @@ type OpenshiftControllerConfig struct {
 	DeploymentConfigControllerConfig  DeploymentConfigControllerConfig
 	DeploymentTriggerControllerConfig DeploymentTriggerControllerConfig
 
-	ImageTriggerControllerConfig ImageTriggerControllerConfig
-	ImageImportControllerConfig  ImageImportControllerConfig
+	ImageTriggerControllerConfig         ImageTriggerControllerConfig
+	ImageSignatureImportControllerConfig ImageSignatureImportControllerConfig
+	ImageImportControllerConfig          ImageImportControllerConfig
 
 	ServiceServingCertsControllerOptions ServiceServingCertsControllerOptions
 
@@ -80,6 +81,7 @@ func (c *OpenshiftControllerConfig) GetControllerInitializers() (map[string]Init
 
 	ret["openshift.io/image-trigger"] = c.ImageTriggerControllerConfig.RunController
 	ret["openshift.io/image-import"] = c.ImageImportControllerConfig.RunController
+	ret["openshift.io/image-signature-import"] = c.ImageSignatureImportControllerConfig.RunController
 
 	ret["openshift.io/templateinstance"] = RunTemplateInstanceController
 
@@ -203,6 +205,11 @@ func BuildOpenshiftControllerConfig(options configapi.MasterConfig) (*OpenshiftC
 		DisableScheduledImport:                     options.ImagePolicyConfig.DisableScheduledImport,
 		ScheduledImageImportMinimumIntervalSeconds: options.ImagePolicyConfig.ScheduledImageImportMinimumIntervalSeconds,
 	}
+	ret.ImageSignatureImportControllerConfig = ImageSignatureImportControllerConfig{
+		ResyncPeriod:          10 * time.Minute,
+		SignatureFetchTimeout: 1 * time.Minute,
+		SignatureImportLimit:  3,
+	}
 
 	ret.ServiceServingCertsControllerOptions = ServiceServingCertsControllerOptions{
 		Signer: options.ControllerConfig.ServiceServingCert.Signer,

pkg/cmd/server/origin/controller/image.go

@@ -1,6 +1,7 @@
 package controller
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -18,6 +19,7 @@ import (
 	buildclient "github.com/openshift/origin/pkg/build/client"
 	"github.com/openshift/origin/pkg/cmd/server/bootstrappolicy"
 	imagecontroller "github.com/openshift/origin/pkg/image/controller"
+	imagesignaturecontroller "github.com/openshift/origin/pkg/image/controller/signature"
 	imagetriggercontroller "github.com/openshift/origin/pkg/image/controller/trigger"
 	triggerannotations "github.com/openshift/origin/pkg/image/trigger/annotations"
 	triggerbuildconfigs "github.com/openshift/origin/pkg/image/trigger/buildconfigs"
@@ -147,6 +149,25 @@ func (u podSpecUpdater) Update(obj runtime.Object) error {
 	}
 }
 
+type ImageSignatureImportControllerConfig struct {
+	ResyncPeriod          time.Duration
+	SignatureFetchTimeout time.Duration
+	SignatureImportLimit  int
+}
+
+func (c *ImageSignatureImportControllerConfig) RunController(ctx ControllerContext) (bool, error) {
+	controller := imagesignaturecontroller.NewSignatureImportController(
+		context.Background(),
+		ctx.ClientBuilder.OpenshiftInternalImageClientOrDie(bootstrappolicy.InfraImageImportControllerServiceAccountName),
+		ctx.ImageInformers.Image().InternalVersion().Images(),
+		c.ResyncPeriod,
+		c.SignatureFetchTimeout,
+		c.SignatureImportLimit,
+	)
+	go controller.Run(5, ctx.Stop)
+	return true, nil
+}
+
 type ImageImportControllerConfig struct {
 	MaxScheduledImageImportsPerMinute          int
 	ScheduledImageImportMinimumIntervalSeconds int

pkg/image/controller/signature/container_image_downloader.go

@@ -0,0 +1,62 @@
+package signature
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"time"
+
+	"github.com/containers/image/docker"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	imageapi "github.com/openshift/origin/pkg/image/apis/image"
+)
+
+type containerImageSignatureDownloader struct {
+	ctx     context.Context
+	timeout time.Duration
+}
+
+func NewContainerImageSignatureDownloader(ctx context.Context, timeout time.Duration) SignatureDownloader {
+	return &containerImageSignatureDownloader{
+		ctx:     ctx,
+		timeout: timeout,
+	}
+}
+
+func (s *containerImageSignatureDownloader) DownloadImageSignatures(image *imageapi.Image) ([]imageapi.ImageSignature, error) {
+	reference, err := docker.ParseReference("//" + image.DockerImageReference)
+	if err != nil {
+		return nil, err
+	}
+	source, err := reference.NewImageSource(nil, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer source.Close()
+
+	ctx, cancel := context.WithTimeout(s.ctx, s.timeout)
+	defer cancel()
+
+	signatures, err := source.GetSignatures(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	ret := []imageapi.ImageSignature{}
+	for _, blob := range signatures {
+		sig := imageapi.ImageSignature{Type: imageapi.ImageSignatureTypeAtomicImageV1}
+		// This will use the name of the image (sha256:xxxx) and the SHA256 of the
+		// signature itself as the signature name has to be unique for each
+		// signature.
+		sig.Name = imageapi.JoinImageStreamImage(image.Name, fmt.Sprintf("%x", sha256.Sum256(blob)))
+		sig.Content = blob
+		sig.Annotations = map[string]string{
+			SignatureManagedAnnotation: "true",
+		}
+		sig.CreationTimestamp = metav1.Now()
+		ret = append(ret, sig)
+	}
+	return ret, nil
+}

pkg/image/controller/signature/signature_import_controller.go

@@ -0,0 +1,192 @@
+package signature
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/golang/glog"
+
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	kapi "k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/controller"
+
+	imageapi "github.com/openshift/origin/pkg/image/apis/image"
+	informers "github.com/openshift/origin/pkg/image/generated/informers/internalversion/image/internalversion"
+	imageclient "github.com/openshift/origin/pkg/image/generated/internalclientset"
+	imagelister "github.com/openshift/origin/pkg/image/generated/listers/image/internalversion"
+)
+
+const (
+	// SignatureManagedAnnotation marks signatures that were imported by this
+	// controller.
+	SignatureManagedAnnotation = "image.openshift.io/managed-signature"
+)
+
+type SignatureDownloader interface {
+	DownloadImageSignatures(*imageapi.Image) ([]imageapi.ImageSignature, error)
+}
+
+type SignatureImportController struct {
+	imageClient imageclient.Interface
+	imageLister imagelister.ImageLister
+
+	imageHasSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+
+	// signatureImportLimit limits amount of signatures we will import.
+	// By default this is set to 3 signatures.
+	signatureImportLimit int
+
+	fetcher SignatureDownloader
+}
+
+func NewSignatureImportController(ctx context.Context, imageClient imageclient.Interface, imageInformer informers.ImageInformer, resyncInterval, fetchTimeout time.Duration, limit int) *SignatureImportController {
+	controller := &SignatureImportController{
+		queue:                workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		imageClient:          imageClient,
+		imageLister:          imageInformer.Lister(),
+		imageHasSynced:       imageInformer.Informer().HasSynced,
+		signatureImportLimit: limit,
+	}
+	controller.fetcher = NewContainerImageSignatureDownloader(ctx, fetchTimeout)
+
+	imageInformer.Informer().AddEventHandlerWithResyncPeriod(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			image := obj.(*imageapi.Image)
+			glog.V(4).Infof("Adding image %s", image.Name)
+			controller.enqueueImage(obj)
+		},
+		UpdateFunc: func(old, cur interface{}) {
+			image := cur.(*imageapi.Image)
+			glog.V(4).Infof("Updating image %s", image.Name)
+			controller.enqueueImage(cur)
+		},
+	}, resyncInterval)
+
+	return controller
+}
+
+func (s *SignatureImportController) Run(workers int, stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer s.queue.ShutDown()
+
+	if !cache.WaitForCacheSync(stopCh, s.imageHasSynced) {
+		return
+	}
+
+	glog.V(5).Infof("Starting workers")
+	for i := 0; i < workers; i++ {
+		go wait.Until(s.worker, time.Second, stopCh)
+	}
+	<-stopCh
+	glog.V(1).Infof("Shutting down")
+
+}
+
+func (s *SignatureImportController) worker() {
+	for {
+		if !s.work() {
+			return
+		}
+	}
+}
+
+// work returns true if the worker thread should continue
+func (s *SignatureImportController) work() bool {
+	key, quit := s.queue.Get()
+	if quit {
+		return false
+	}
+	defer s.queue.Done(key)
+
+	err := s.syncImageSignatures(key.(string))
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("error syncing image %s, it will be retried: %v", key.(string), err))
+		s.queue.AddRateLimited(key)
+		return true
+	}
+
+	s.queue.Forget(key)
+	return true
+}
+
+func (s *SignatureImportController) enqueueImage(obj interface{}) {
+	_, ok := obj.(*imageapi.Image)
+	if !ok {
+		return
+	}
+	key, err := controller.KeyFunc(obj)
+	if err != nil {
+		glog.Errorf("Couldn't get key for object %+v: %v", obj, err)
+		return
+	}
+	s.queue.Add(key)
+}
+
+func (s *SignatureImportController) syncImageSignatures(key string) error {
+	glog.V(4).Infof("Initiating download of signatures for %s", key)
+	image, err := s.imageLister.Get(key)
+	if err != nil {
+		glog.V(4).Infof("Unable to get image %v: %v", key, err)
+		return err
+	}
+
+	currentSignatures, err := s.fetcher.DownloadImageSignatures(image)
+	if err != nil {
+		glog.V(4).Infof("Failed to fetch image %s signatures: %v", image.Name, err)
+		return err
+	}
+
+	// Having no signatures means no-op (we don't remove stored signatures when
+	// the sig-store no longer have them).
+	if len(currentSignatures) == 0 {
+		glog.V(4).Infof("No signatures dowloaded for %s", image.Name)
+		return nil
+	}
+
+	t, err := kapi.Scheme.DeepCopy(image)
+	if err != nil {
+		return err
+	}
+	newImage := t.(*imageapi.Image)
+
+	shouldUpdate := false
+
+	// Only add new signatures, do not override existing stored signatures as that
+	// can void their verification status.
+	for _, c := range currentSignatures {
+		found := false
+		for _, s := range newImage.Signatures {
+			if s.Name == c.Name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			newImage.Signatures = append(newImage.Signatures, c)
+			shouldUpdate = true
+		}
+	}
+
+	if len(newImage.Signatures) > s.signatureImportLimit {
+		glog.V(2).Infof("Image %s reached signature limit (max:%d, want:%d)", newImage.Name, s.signatureImportLimit, len(newImage.Signatures))
+		return nil
+	}
+
+	// Avoid unnecessary updates to images.
+	if !shouldUpdate {
+		return nil
+	}
+	glog.V(4).Infof("Image %s now has %d signatures", newImage.Name, len(newImage.Signatures))
+
+	if _, err := s.imageClient.Image().Images().Update(newImage); err != nil {
+		return err
+	}
+
+	return nil
+}