package gcs import ( "context" "encoding/hex" "fmt" "io" "os" "path/filepath" "strings" "cloud.google.com/go/storage" "github.com/aptly-dev/aptly/aptly" "github.com/aptly-dev/aptly/utils" "github.com/pkg/errors" "github.com/rs/zerolog/log" "google.golang.org/api/googleapi" "google.golang.org/api/iterator" "google.golang.org/api/option" ) // PublishedStorage abstracts published files hosted on GCS. type PublishedStorage struct { client *storage.Client bucket *storage.BucketHandle bucketName string prefix string acl string storageClass string encryptionKey string disableMultiDel bool debug bool pathCache map[string]string } var ( _ aptly.PublishedStorage = (*PublishedStorage)(nil) ) // NewPublishedStorage creates a GCS-backed published storage. func NewPublishedStorage(bucket, prefix, credentialsFile, serviceAccountJSON, project, endpoint, defaultACL, storageClass, encryptionKey string, disableMultiDel, debug bool) (*PublishedStorage, error) { ctx := context.TODO() opts := make([]option.ClientOption, 0, 4) if endpoint != "" { opts = append(opts, option.WithEndpoint(endpoint), option.WithoutAuthentication()) } else if credentialsFile != "" { opts = append(opts, option.WithAuthCredentialsFile(option.ServiceAccount, credentialsFile)) } else if serviceAccountJSON != "" { opts = append(opts, option.WithAuthCredentialsJSON(option.ServiceAccount, []byte(serviceAccountJSON))) } if project != "" { opts = append(opts, option.WithQuotaProject(project)) } // When pointing at a non-production endpoint (an explicit override or the // STORAGE_EMULATOR_HOST hook the GCS Go client honours natively), force // JSON reads. The default XML download API uses virtual-host-style URLs // (https://.storage.googleapis.com/...) that emulators and most // dev/staging endpoints can't serve under a single listener; the JSON API // uses path-based URLs that work the same against real GCS or an emulator. if endpoint != "" || os.Getenv("STORAGE_EMULATOR_HOST") != "" { opts = append(opts, storage.WithJSONReads()) } client, err := storage.NewClient(ctx, opts...) if err != nil { return nil, err } result := &PublishedStorage{ client: client, bucket: client.Bucket(bucket), bucketName: bucket, prefix: prefix, acl: defaultACL, storageClass: storageClass, encryptionKey: encryptionKey, disableMultiDel: disableMultiDel, debug: debug, } return result, nil } func (g *PublishedStorage) String() string { return fmt.Sprintf("GCS: %s/%s", g.bucketName, g.prefix) } // MkDir creates directory recursively under public path. func (g *PublishedStorage) MkDir(_ string) error { // no-op for GCS return nil } func (g *PublishedStorage) objectPath(path string) string { return filepath.Join(g.prefix, path) } func (g *PublishedStorage) objectHandle(path string) *storage.ObjectHandle { obj := g.bucket.Object(g.objectPath(path)) if g.encryptionKey != "" { obj = obj.Key([]byte(g.encryptionKey)) } return obj } // PutFile puts file into published storage at specified path. func (g *PublishedStorage) PutFile(path string, sourceFilename string) error { source, err := os.Open(sourceFilename) if err != nil { return err } defer func() { _ = source.Close() }() if g.debug { log.Debug().Msgf("GCS: PutFile '%s'", path) } err = g.putFile(path, source, "") if err != nil { return errors.Wrap(err, fmt.Sprintf("error uploading %s to %s", sourceFilename, g)) } return nil } func (g *PublishedStorage) applyACL(obj *storage.ObjectHandle) error { switch g.acl { case "", "none", "private": return nil case "public-read": return obj.ACL().Set(context.TODO(), storage.AllUsers, storage.RoleReader) default: return fmt.Errorf("unsupported GCS ACL value: %s", g.acl) } } func (g *PublishedStorage) putFile(path string, source io.Reader, sourceMD5 string) error { obj := g.objectHandle(path) writer := obj.NewWriter(context.TODO()) if g.storageClass != "" { writer.StorageClass = g.storageClass } if sourceMD5 != "" { writer.Metadata = map[string]string{"Md5": sourceMD5} } if _, err := io.Copy(writer, source); err != nil { _ = writer.Close() return err } if err := writer.Close(); err != nil { return err } return g.applyACL(obj) } func (g *PublishedStorage) getMD5(path string) (string, error) { attrs, err := g.objectHandle(path).Attrs(context.TODO()) if err != nil { return "", err } if attrs.Metadata != nil { if md5, ok := attrs.Metadata["Md5"]; ok && md5 != "" { return strings.ToLower(md5), nil } } return strings.ToLower(hex.EncodeToString(attrs.MD5)), nil } // Remove removes single file under public path. func (g *PublishedStorage) Remove(path string) error { if g.debug { log.Debug().Msgf("GCS: Remove '%s'", path) } err := g.objectHandle(path).Delete(context.TODO()) if err != nil { if err == storage.ErrObjectNotExist { return nil } var apiErr *googleapi.Error if errors.As(err, &apiErr) && apiErr.Code == 404 { return nil } return errors.Wrap(err, fmt.Sprintf("error deleting %s from %s", path, g)) } delete(g.pathCache, path) return nil } // RemoveDirs removes directory structure under public path. func (g *PublishedStorage) RemoveDirs(path string, _ aptly.Progress) error { filelist, _, err := g.internalFilelist(path) if err != nil { return err } if g.debug { log.Debug().Msgf("GCS: RemoveDirs '%s'", path) } for _, file := range filelist { objPath := filepath.Join(path, file) if err := g.Remove(objPath); err != nil { return err } } _ = g.disableMultiDel return nil } // LinkFromPool links package file from pool to dist's pool location. func (g *PublishedStorage) LinkFromPool(publishedPrefix, publishedRelPath, fileName string, sourcePool aptly.PackagePool, sourcePath string, sourceChecksums utils.ChecksumInfo, force bool) error { publishedDirectory := filepath.Join(publishedPrefix, publishedRelPath) relPath := filepath.Join(publishedDirectory, fileName) poolPath := filepath.Join(g.prefix, relPath) if g.pathCache == nil { paths, md5s, err := g.internalFilelist(filepath.Join(publishedPrefix, "pool")) if err != nil { return errors.Wrap(err, "error caching paths under prefix") } g.pathCache = make(map[string]string, len(paths)) for i := range paths { g.pathCache[filepath.Join(publishedPrefix, "pool", paths[i])] = md5s[i] } } destinationMD5, exists := g.pathCache[relPath] sourceMD5 := strings.ToLower(sourceChecksums.MD5) if exists { if sourceMD5 == "" { return fmt.Errorf("unable to compare object, MD5 checksum missing") } if len(destinationMD5) != 32 { var err error destinationMD5, err = g.getMD5(relPath) if err != nil { return errors.Wrap(err, fmt.Sprintf("error verifying MD5 for %s: %s", g, poolPath)) } g.pathCache[relPath] = destinationMD5 } if destinationMD5 == sourceMD5 { return nil } if !force { return fmt.Errorf("error putting file to %s: file already exists and is different: %s", poolPath, g) } } source, err := sourcePool.Open(sourcePath) if err != nil { return err } defer func() { _ = source.Close() }() if g.debug { log.Debug().Msgf("GCS: LinkFromPool '%s'", relPath) } err = g.putFile(relPath, source, sourceMD5) if err != nil { return errors.Wrap(err, fmt.Sprintf("error uploading %s to %s: %s", sourcePath, g, poolPath)) } g.pathCache[relPath] = sourceMD5 return nil } // Filelist returns list of files under prefix. func (g *PublishedStorage) Filelist(prefix string) ([]string, error) { paths, _, err := g.internalFilelist(prefix) return paths, err } func (g *PublishedStorage) internalFilelist(prefix string) ([]string, []string, error) { paths := make([]string, 0, 1024) md5s := make([]string, 0, 1024) fullPrefix := filepath.Join(g.prefix, prefix) if fullPrefix != "" { fullPrefix += "/" } it := g.bucket.Objects(context.TODO(), &storage.Query{Prefix: fullPrefix}) for { attrs, err := it.Next() if err == iterator.Done { break } if err != nil { return nil, nil, errors.WithMessagef(err, "error listing under prefix %s in %s", fullPrefix, g) } path := attrs.Name if fullPrefix != "" { path = strings.TrimPrefix(path, fullPrefix) } paths = append(paths, path) if attrs.Metadata != nil { if md5, ok := attrs.Metadata["Md5"]; ok && md5 != "" { md5s = append(md5s, strings.ToLower(md5)) continue } } md5s = append(md5s, strings.ToLower(hex.EncodeToString(attrs.MD5))) } return paths, md5s, nil } // RenameFile renames (moves) file. func (g *PublishedStorage) RenameFile(oldName, newName string) error { src := g.objectHandle(oldName) dst := g.objectHandle(newName) if g.debug { log.Debug().Msgf("GCS: RenameFile %s -> %s", oldName, newName) } _, err := dst.CopierFrom(src).Run(context.TODO()) if err != nil { return fmt.Errorf("error copying %s -> %s in %s: %s", oldName, newName, g, err) } err = g.applyACL(dst) if err != nil { return err } return g.Remove(oldName) } // SymLink creates a copy of src file and stores link information in metadata. func (g *PublishedStorage) SymLink(src string, dst string) error { source := g.objectHandle(src) dest := g.objectHandle(dst) if g.debug { log.Debug().Msgf("GCS: SymLink %s -> %s", src, dst) } _, err := dest.CopierFrom(source).Run(context.TODO()) if err != nil { return fmt.Errorf("error symlinking %s -> %s in %s: %s", src, dst, g, err) } _, err = dest.Update(context.TODO(), storage.ObjectAttrsToUpdate{ Metadata: map[string]string{"SymLink": src}, }) if err != nil { return fmt.Errorf("error updating symlink metadata %s -> %s in %s: %s", src, dst, g, err) } return g.applyACL(dest) } // HardLink uses symlink functionality as hard links do not exist on object stores. func (g *PublishedStorage) HardLink(src string, dst string) error { if g.debug { log.Debug().Msgf("GCS: HardLink %s -> %s", src, dst) } return g.SymLink(src, dst) } // FileExists returns true if path exists. func (g *PublishedStorage) FileExists(path string) (bool, error) { _, err := g.objectHandle(path).Attrs(context.TODO()) if err != nil { if err == storage.ErrObjectNotExist { return false, nil } var apiErr *googleapi.Error if errors.As(err, &apiErr) && apiErr.Code == 404 { return false, nil } return false, err } return true, nil } // ReadLink returns symbolic link target from metadata. func (g *PublishedStorage) ReadLink(path string) (string, error) { attrs, err := g.objectHandle(path).Attrs(context.TODO()) if err != nil { return "", err } return attrs.Metadata["SymLink"], nil }