219 lines
		
	
	
	
		
			5.9 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			219 lines
		
	
	
	
		
			5.9 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
// Copyright 2019 Google LLC
 | 
						|
//
 | 
						|
// 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
 | 
						|
//
 | 
						|
//     https://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.
 | 
						|
 | 
						|
// Google Cloud Storage backend for Nixery.
 | 
						|
package storage
 | 
						|
 | 
						|
import (
 | 
						|
	"context"
 | 
						|
	"fmt"
 | 
						|
	"io"
 | 
						|
	"io/ioutil"
 | 
						|
	"net/http"
 | 
						|
	"net/url"
 | 
						|
	"os"
 | 
						|
	"time"
 | 
						|
 | 
						|
	"cloud.google.com/go/storage"
 | 
						|
	log "github.com/sirupsen/logrus"
 | 
						|
	"golang.org/x/oauth2/google"
 | 
						|
)
 | 
						|
 | 
						|
// HTTP client to use for direct calls to APIs that are not part of the SDK
 | 
						|
var client = &http.Client{}
 | 
						|
 | 
						|
// API scope needed for renaming objects in GCS
 | 
						|
const gcsScope = "https://www.googleapis.com/auth/devstorage.read_write"
 | 
						|
 | 
						|
type GCSBackend struct {
 | 
						|
	bucket  string
 | 
						|
	handle  *storage.BucketHandle
 | 
						|
	signing *storage.SignedURLOptions
 | 
						|
}
 | 
						|
 | 
						|
// Constructs a new GCS bucket backend based on the configured
 | 
						|
// environment variables.
 | 
						|
func NewGCSBackend() (*GCSBackend, error) {
 | 
						|
	bucket := os.Getenv("GCS_BUCKET")
 | 
						|
	if bucket == "" {
 | 
						|
		return nil, fmt.Errorf("GCS_BUCKET must be configured for GCS usage")
 | 
						|
	}
 | 
						|
 | 
						|
	ctx := context.Background()
 | 
						|
	client, err := storage.NewClient(ctx)
 | 
						|
	if err != nil {
 | 
						|
		log.WithError(err).Fatal("failed to set up Cloud Storage client")
 | 
						|
	}
 | 
						|
 | 
						|
	handle := client.Bucket(bucket)
 | 
						|
 | 
						|
	if _, err := handle.Attrs(ctx); err != nil {
 | 
						|
		log.WithError(err).WithField("bucket", bucket).Error("could not access configured bucket")
 | 
						|
		return nil, err
 | 
						|
	}
 | 
						|
 | 
						|
	signing, err := signingOptsFromEnv()
 | 
						|
	if err != nil {
 | 
						|
		log.WithError(err).Error("failed to configure GCS bucket signing")
 | 
						|
		return nil, err
 | 
						|
	}
 | 
						|
 | 
						|
	return &GCSBackend{
 | 
						|
		bucket:  bucket,
 | 
						|
		handle:  handle,
 | 
						|
		signing: signing,
 | 
						|
	}, nil
 | 
						|
}
 | 
						|
 | 
						|
func (b *GCSBackend) Name() string {
 | 
						|
	return "Google Cloud Storage (" + b.bucket + ")"
 | 
						|
}
 | 
						|
 | 
						|
func (b *GCSBackend) Persist(ctx context.Context, path string, f Persister) (string, int64, error) {
 | 
						|
	obj := b.handle.Object(path)
 | 
						|
	w := obj.NewWriter(ctx)
 | 
						|
 | 
						|
	hash, size, err := f(w)
 | 
						|
	if err != nil {
 | 
						|
		log.WithError(err).WithField("path", path).Error("failed to upload to GCS")
 | 
						|
		return hash, size, err
 | 
						|
	}
 | 
						|
 | 
						|
	return hash, size, w.Close()
 | 
						|
}
 | 
						|
 | 
						|
func (b *GCSBackend) Fetch(ctx context.Context, path string) (io.ReadCloser, error) {
 | 
						|
	obj := b.handle.Object(path)
 | 
						|
 | 
						|
	// Probe whether the file exists before trying to fetch it
 | 
						|
	_, err := obj.Attrs(ctx)
 | 
						|
	if err != nil {
 | 
						|
		return nil, err
 | 
						|
	}
 | 
						|
 | 
						|
	return obj.NewReader(ctx)
 | 
						|
}
 | 
						|
 | 
						|
// renameObject renames an object in the specified Cloud Storage
 | 
						|
// bucket.
 | 
						|
//
 | 
						|
// The Go API for Cloud Storage does not support renaming objects, but
 | 
						|
// the HTTP API does. The code below makes the relevant call manually.
 | 
						|
func (b *GCSBackend) Move(ctx context.Context, old, new string) error {
 | 
						|
	creds, err := google.FindDefaultCredentials(ctx, gcsScope)
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	token, err := creds.TokenSource.Token()
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	// as per https://cloud.google.com/storage/docs/renaming-copying-moving-objects#rename
 | 
						|
	url := fmt.Sprintf(
 | 
						|
		"https://www.googleapis.com/storage/v1/b/%s/o/%s/rewriteTo/b/%s/o/%s",
 | 
						|
		url.PathEscape(b.bucket), url.PathEscape(old),
 | 
						|
		url.PathEscape(b.bucket), url.PathEscape(new),
 | 
						|
	)
 | 
						|
 | 
						|
	req, err := http.NewRequest("POST", url, nil)
 | 
						|
	req.Header.Add("Authorization", "Bearer "+token.AccessToken)
 | 
						|
	_, err = client.Do(req)
 | 
						|
	if err != nil {
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	// It seems that 'rewriteTo' copies objects instead of
 | 
						|
	// renaming/moving them, hence a deletion call afterwards is
 | 
						|
	// required.
 | 
						|
	if err = b.handle.Object(old).Delete(ctx); err != nil {
 | 
						|
		log.WithError(err).WithFields(log.Fields{
 | 
						|
			"new": new,
 | 
						|
			"old": old,
 | 
						|
		}).Warn("failed to delete renamed object")
 | 
						|
 | 
						|
		// this error should not break renaming and is not returned
 | 
						|
	}
 | 
						|
 | 
						|
	return nil
 | 
						|
}
 | 
						|
 | 
						|
func (b *GCSBackend) ServeLayer(digest string, r *http.Request, w http.ResponseWriter) error {
 | 
						|
	url, err := b.constructLayerUrl(digest)
 | 
						|
	if err != nil {
 | 
						|
		log.WithError(err).WithFields(log.Fields{
 | 
						|
			"layer":  digest,
 | 
						|
			"bucket": b.bucket,
 | 
						|
		}).Error("failed to sign GCS URL")
 | 
						|
 | 
						|
		return err
 | 
						|
	}
 | 
						|
 | 
						|
	log.WithField("layer", digest).Info("redirecting layer request to GCS bucket")
 | 
						|
 | 
						|
	w.Header().Set("Location", url)
 | 
						|
	w.WriteHeader(303)
 | 
						|
	return nil
 | 
						|
}
 | 
						|
 | 
						|
// Configure GCS URL signing in the presence of a service account key
 | 
						|
// (toggled if the user has set GOOGLE_APPLICATION_CREDENTIALS).
 | 
						|
func signingOptsFromEnv() (*storage.SignedURLOptions, error) {
 | 
						|
	path := os.Getenv("GOOGLE_APPLICATION_CREDENTIALS")
 | 
						|
	if path == "" {
 | 
						|
		// No credentials configured -> no URL signing
 | 
						|
		return nil, nil
 | 
						|
	}
 | 
						|
 | 
						|
	key, err := ioutil.ReadFile(path)
 | 
						|
	if err != nil {
 | 
						|
		return nil, fmt.Errorf("failed to read service account key: %s", err)
 | 
						|
	}
 | 
						|
 | 
						|
	conf, err := google.JWTConfigFromJSON(key)
 | 
						|
	if err != nil {
 | 
						|
		return nil, fmt.Errorf("failed to parse service account key: %s", err)
 | 
						|
	}
 | 
						|
 | 
						|
	log.WithField("account", conf.Email).Info("GCS URL signing enabled")
 | 
						|
 | 
						|
	return &storage.SignedURLOptions{
 | 
						|
		Scheme:         storage.SigningSchemeV4,
 | 
						|
		GoogleAccessID: conf.Email,
 | 
						|
		PrivateKey:     conf.PrivateKey,
 | 
						|
		Method:         "GET",
 | 
						|
	}, nil
 | 
						|
}
 | 
						|
 | 
						|
// layerRedirect constructs the public URL of the layer object in the Cloud
 | 
						|
// Storage bucket, signs it and redirects the user there.
 | 
						|
//
 | 
						|
// Signing the URL allows unauthenticated clients to retrieve objects from the
 | 
						|
// bucket.
 | 
						|
//
 | 
						|
// The Docker client is known to follow redirects, but this might not be true
 | 
						|
// for all other registry clients.
 | 
						|
func (b *GCSBackend) constructLayerUrl(digest string) (string, error) {
 | 
						|
	log.WithField("layer", digest).Info("redirecting layer request to bucket")
 | 
						|
	object := "layers/" + digest
 | 
						|
 | 
						|
	if b.signing != nil {
 | 
						|
		opts := *b.signing
 | 
						|
		opts.Expires = time.Now().Add(5 * time.Minute)
 | 
						|
		return storage.SignedURL(b.bucket, object, &opts)
 | 
						|
	} else {
 | 
						|
		return ("https://storage.googleapis.com/" + b.bucket + "/" + object), nil
 | 
						|
	}
 | 
						|
}
 |