From 0477660f8f13c00e0d10fb5af5d20fb8843d683d Mon Sep 17 00:00:00 2001 From: "Anthony A." Date: Thu, 11 Jun 2026 13:35:19 -0400 Subject: [PATCH 1/3] Add GCS storage backend with Workload Identity support Register gocloud.dev/blob/gcsblob so gs:// URLs are accepted as a storage backend. Authentication uses Application Default Credentials, which makes GKE Workload Identity work out of the box; signed URLs (direct_serve) fall back to the IAM Credentials signBlob API when no private key is available. Co-Authored-By: Claude Opus 4.7 --- README.md | 53 ++++++++++++++++++++++++++++++++++++- config.example.yaml | 13 ++++++++- internal/config/config.go | 17 +++++++++++- internal/storage/blob.go | 4 +++ internal/storage/storage.go | 3 +++ 5 files changed, 87 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 320a737..ba70b42 100644 --- a/README.md +++ b/README.md @@ -449,7 +449,7 @@ The proxy can be configured via: -config string Path to configuration file -listen string Address to listen on (default ":8080") -base-url string Public URL of this proxy (default "http://localhost:8080") --storage-url string Storage URL (file:// or s3://) +-storage-url string Storage URL (file://, s3://, gs://, azblob://) -storage-path string Path to artifact storage directory (deprecated, use -storage-url) -database-driver string Database driver: sqlite or postgres (default "sqlite") -database-path string Path to SQLite database file (default "./cache/proxy.db") @@ -549,6 +549,57 @@ storage: Set credentials via standard AWS environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_REGION`). +### Google Cloud Storage + +The proxy can store cached artifacts in a GCS bucket using the `gs://` URL scheme. + +```yaml +storage: + url: "gs://my-bucket-name" +``` + +Authentication uses [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials), which means no credentials need to be embedded in the config or environment. Supported sources, in order: + +- **GKE Workload Identity** — bind the Kubernetes service account running the proxy to a Google service account that has `roles/storage.objectAdmin` on the bucket. The proxy will use the workload's token automatically. +- **Attached service account** on GCE, Cloud Run, Cloud Functions, etc. +- **`GOOGLE_APPLICATION_CREDENTIALS`** environment variable pointing at a service account JSON key file. +- **`gcloud auth application-default login`** for local development. + +#### GKE Workload Identity setup + +```bash +# 1. Create a Google service account +gcloud iam service-accounts create git-pkgs-proxy \ + --project=PROJECT_ID + +# 2. Grant it access to the bucket +gsutil iam ch \ + serviceAccount:git-pkgs-proxy@PROJECT_ID.iam.gserviceaccount.com:objectAdmin \ + gs://my-bucket-name + +# 3. Bind the Kubernetes service account to it +gcloud iam service-accounts add-iam-policy-binding \ + git-pkgs-proxy@PROJECT_ID.iam.gserviceaccount.com \ + --role=roles/iam.workloadIdentityUser \ + --member="serviceAccount:PROJECT_ID.svc.id.goog[NAMESPACE/KSA_NAME]" + +# 4. Annotate the Kubernetes service account +kubectl annotate serviceaccount KSA_NAME \ + --namespace=NAMESPACE \ + iam.gke.io/gcp-service-account=git-pkgs-proxy@PROJECT_ID.iam.gserviceaccount.com +``` + +#### Direct serve (signed URLs) with Workload Identity + +When `direct_serve: true` is enabled, the proxy issues HTTP 302 redirects to presigned GCS URLs. Because Workload Identity provides no private key, the gcsblob driver falls back to the [IAM Credentials `signBlob` API](https://cloud.google.com/iam/docs/reference/credentials/rest/v1/projects.serviceAccounts/signBlob). For this to work, grant the service account the token-creator role on itself: + +```bash +gcloud iam service-accounts add-iam-policy-binding \ + git-pkgs-proxy@PROJECT_ID.iam.gserviceaccount.com \ + --role=roles/iam.serviceAccountTokenCreator \ + --member="serviceAccount:git-pkgs-proxy@PROJECT_ID.iam.gserviceaccount.com" +``` + ## CLI Commands ### serve (default) diff --git a/config.example.yaml b/config.example.yaml index 1df95b3..b4470a8 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -27,9 +27,20 @@ storage: # - file:///path/to/dir - Local filesystem (default) # - s3://bucket-name - Amazon S3 # - s3://bucket?endpoint=http://localhost:9000 - S3-compatible (MinIO) + # - gs://bucket-name - Google Cloud Storage + # - azblob://container-name - Azure Blob Storage # # For S3, configure credentials via environment variables: # AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION + # + # For GCS, authentication uses Application Default Credentials. On GKE with + # Workload Identity, bind the Kubernetes service account to a Google service + # account that has roles/storage.objectAdmin on the bucket. No extra config + # is needed in this file. For local development, run: + # gcloud auth application-default login + # If direct_serve is enabled, the service account also needs + # roles/iam.serviceAccountTokenCreator on itself so the IAM Credentials + # signBlob API can sign URLs without a private key. url: "" # Local filesystem path (used when url is empty) @@ -42,7 +53,7 @@ storage: max_size: "" # Redirect cached artifact downloads to presigned storage URLs (HTTP 302) - # instead of streaming through the proxy. Only effective for S3 and Azure. + # instead of streaming through the proxy. Only effective for S3, GCS, and Azure. # Leave disabled if clients reach the proxy through an authenticating gateway, # since presigned URLs bypass it. direct_serve: false diff --git a/internal/config/config.go b/internal/config/config.go index a3dfbc6..89c2616 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,10 +24,23 @@ // storage: // url: "s3://bucket?endpoint=http://localhost:9000" // +// Google Cloud Storage: +// +// storage: +// url: "gs://bucket-name" +// // For S3, configure credentials via AWS environment variables: // // AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION // +// For GCS, authentication uses Application Default Credentials. This works +// transparently on GKE with Workload Identity, GCE/Cloud Run with attached +// service accounts, or locally via `gcloud auth application-default login`. +// When the proxy is signing URLs (direct_serve) without a private key, +// gcsblob automatically falls back to the IAM Credentials signBlob API, +// which requires the service account to hold roles/iam.serviceAccountTokenCreator +// on itself. +// // Database Configuration: // // The proxy supports two database backends: @@ -181,6 +194,8 @@ type StorageConfig struct { // - file:///path/to/dir - Local filesystem (default) // - s3://bucket-name - Amazon S3 // - s3://bucket?endpoint=http://localhost:9000 - S3-compatible (MinIO) + // - gs://bucket-name - Google Cloud Storage (Workload Identity supported) + // - azblob://container-name - Azure Blob Storage // If empty, defaults to file:// with the Path value. URL string `json:"url" yaml:"url"` @@ -197,7 +212,7 @@ type StorageConfig struct { // DirectServe enables redirecting cached artifact downloads to presigned // storage URLs (HTTP 302) instead of streaming bytes through the proxy. - // Only effective for backends that support URL signing (S3, Azure). + // Only effective for backends that support URL signing (S3, GCS, Azure). DirectServe bool `json:"direct_serve" yaml:"direct_serve"` // DirectServeTTL is how long presigned URLs remain valid. diff --git a/internal/storage/blob.go b/internal/storage/blob.go index 67e91d0..98b17e8 100644 --- a/internal/storage/blob.go +++ b/internal/storage/blob.go @@ -16,6 +16,7 @@ import ( "gocloud.dev/blob" _ "gocloud.dev/blob/azureblob" _ "gocloud.dev/blob/fileblob" + _ "gocloud.dev/blob/gcsblob" _ "gocloud.dev/blob/s3blob" "gocloud.dev/gcerrors" ) @@ -35,6 +36,9 @@ type Blob struct { // - file:///path/to/dir - Local filesystem storage // - s3://bucket-name - Amazon S3 (uses AWS_* environment variables) // - s3://bucket-name?region=us-east-1&endpoint=http://localhost:9000 - S3-compatible (MinIO, etc.) +// - gs://bucket-name - Google Cloud Storage (uses Application Default Credentials; +// supports Workload Identity on GKE/GCE without any extra configuration) +// - azblob://container-name - Azure Blob Storage // // For local filesystem, the directory is created if it doesn't exist. func OpenBucket(ctx context.Context, urlStr string) (*Blob, error) { diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 3d0be1c..5ff86f2 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -5,6 +5,9 @@ // - file:///path/to/dir - Local filesystem storage // - s3://bucket-name - Amazon S3 // - s3://bucket?endpoint=http://localhost:9000 - S3-compatible (MinIO) +// - gs://bucket-name - Google Cloud Storage (supports GKE Workload Identity +// via Application Default Credentials) +// - azblob://container-name - Azure Blob Storage // // Use OpenBucket to create a storage backend from a URL. package storage From edc18e939d8418ae97a69736e6da807bc0fbfb21 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Wed, 1 Jul 2026 18:56:35 +0100 Subject: [PATCH 2/3] Replace gcsblob with lightweight GCS backend --- internal/storage/blob.go | 60 ++++- internal/storage/gcs.go | 455 +++++++++++++++++++++++++++++++++++ internal/storage/gcs_test.go | 145 +++++++++++ 3 files changed, 657 insertions(+), 3 deletions(-) create mode 100644 internal/storage/gcs.go create mode 100644 internal/storage/gcs_test.go diff --git a/internal/storage/blob.go b/internal/storage/blob.go index 98b17e8..34f6f08 100644 --- a/internal/storage/blob.go +++ b/internal/storage/blob.go @@ -16,7 +16,6 @@ import ( "gocloud.dev/blob" _ "gocloud.dev/blob/azureblob" _ "gocloud.dev/blob/fileblob" - _ "gocloud.dev/blob/gcsblob" _ "gocloud.dev/blob/s3blob" "gocloud.dev/gcerrors" ) @@ -26,8 +25,9 @@ const osWindows = "windows" // Blob implements Storage using gocloud.dev/blob. // Supports local filesystem (file://) and S3 (s3://) URLs. type Blob struct { - bucket *blob.Bucket - url string + bucket *blob.Bucket + backend Storage + url string } // OpenBucket opens a blob bucket from a URL. @@ -42,6 +42,14 @@ type Blob struct { // // For local filesystem, the directory is created if it doesn't exist. func OpenBucket(ctx context.Context, urlStr string) (*Blob, error) { + if strings.HasPrefix(urlStr, "gs://") { + backend, err := OpenGCS(ctx, urlStr) + if err != nil { + return nil, err + } + return &Blob{backend: backend, url: urlStr}, nil + } + // Handle file:// URLs specially to create the directory if strings.HasPrefix(urlStr, "file://") { path := strings.TrimPrefix(urlStr, "file://") @@ -94,6 +102,10 @@ func OpenBucket(ctx context.Context, urlStr string) (*Blob, error) { } func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { + if b.backend != nil { + return b.backend.Store(ctx, path, r) + } + // Compute hash while writing h := sha256.New() tee := io.TeeReader(r, h) @@ -119,6 +131,10 @@ func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, stri } func (b *Blob) Open(ctx context.Context, path string) (io.ReadCloser, error) { + if b.backend != nil { + return b.backend.Open(ctx, path) + } + r, err := b.bucket.NewReader(ctx, path, nil) if err != nil { if isNotExist(err) { @@ -130,6 +146,10 @@ func (b *Blob) Open(ctx context.Context, path string) (io.ReadCloser, error) { } func (b *Blob) Exists(ctx context.Context, path string) (bool, error) { + if b.backend != nil { + return b.backend.Exists(ctx, path) + } + exists, err := b.bucket.Exists(ctx, path) if err != nil { return false, fmt.Errorf("checking existence: %w", err) @@ -138,6 +158,10 @@ func (b *Blob) Exists(ctx context.Context, path string) (bool, error) { } func (b *Blob) Delete(ctx context.Context, path string) error { + if b.backend != nil { + return b.backend.Delete(ctx, path) + } + err := b.bucket.Delete(ctx, path) if err != nil && !isNotExist(err) { return fmt.Errorf("deleting object: %w", err) @@ -146,6 +170,10 @@ func (b *Blob) Delete(ctx context.Context, path string) error { } func (b *Blob) SignedURL(ctx context.Context, path string, expiry time.Duration) (string, error) { + if b.backend != nil { + return b.backend.SignedURL(ctx, path, expiry) + } + url, err := b.bucket.SignedURL(ctx, path, &blob.SignedURLOptions{ Method: http.MethodGet, Expiry: expiry, @@ -160,6 +188,10 @@ func (b *Blob) SignedURL(ctx context.Context, path string, expiry time.Duration) } func (b *Blob) Size(ctx context.Context, path string) (int64, error) { + if b.backend != nil { + return b.backend.Size(ctx, path) + } + attrs, err := b.bucket.Attributes(ctx, path) if err != nil { if isNotExist(err) { @@ -171,6 +203,10 @@ func (b *Blob) Size(ctx context.Context, path string) (int64, error) { } func (b *Blob) UsedSpace(ctx context.Context) (int64, error) { + if b.backend != nil { + return b.backend.UsedSpace(ctx) + } + var total int64 iter := b.bucket.List(nil) @@ -190,6 +226,16 @@ func (b *Blob) UsedSpace(ctx context.Context) (int64, error) { // ListPrefix returns object metadata for keys under a prefix. func (b *Blob) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error) { + if b.backend != nil { + lister, ok := b.backend.(interface { + ListPrefix(context.Context, string) ([]ObjectInfo, error) + }) + if !ok { + return nil, ErrNotFound + } + return lister.ListPrefix(ctx, prefix) + } + iter := b.bucket.List(&blob.ListOptions{Prefix: prefix}) objects := make([]ObjectInfo, 0) @@ -218,10 +264,18 @@ func (b *Blob) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, err } func (b *Blob) Close() error { + if b.backend != nil { + return b.backend.Close() + } + return b.bucket.Close() } func (b *Blob) URL() string { + if b.backend != nil { + return b.backend.URL() + } + return b.url } diff --git a/internal/storage/gcs.go b/internal/storage/gcs.go new file mode 100644 index 0000000..e97e772 --- /dev/null +++ b/internal/storage/gcs.go @@ -0,0 +1,455 @@ +package storage + +import ( + "bytes" + "context" + "crypto" + "crypto/rand" + "crypto/rsa" + "crypto/sha256" + "crypto/x509" + "encoding/base64" + "encoding/hex" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strconv" + "strings" + "time" + + "cloud.google.com/go/compute/metadata" + "golang.org/x/oauth2" + "golang.org/x/oauth2/google" +) + +const ( + gcsScope = "https://www.googleapis.com/auth/cloud-platform" + gcsDefaultHost = "https://storage.googleapis.com" +) + +// GCS implements Storage using the Cloud Storage JSON API. +type GCS struct { + bucket string + url string + client *http.Client + apiBase string + uploadBase string + + accessID string + privateKey []byte + signBytes func(context.Context, []byte) ([]byte, error) +} + +// OpenGCS opens a Google Cloud Storage bucket from a gs:// URL. +func OpenGCS(ctx context.Context, urlStr string) (*GCS, error) { + u, err := url.Parse(urlStr) + if err != nil { + return nil, fmt.Errorf("parsing GCS URL: %w", err) + } + if u.Scheme != "gs" || u.Host == "" { + return nil, fmt.Errorf("invalid GCS URL %q", urlStr) + } + if u.Path != "" && u.Path != "/" { + return nil, fmt.Errorf("GCS URL must name a bucket, got path %q", u.Path) + } + + host := strings.TrimRight(gcsDefaultHost, "/") + client := http.DefaultClient + var accessID string + var privateKey []byte + + if emulator := os.Getenv("STORAGE_EMULATOR_HOST"); emulator != "" { + host = strings.TrimRight(emulator, "/") + if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") { + host = "http://" + host + } + } else { + var credsJSON []byte + var err error + client, credsJSON, err = gcsHTTPClient(ctx) + if err != nil { + return nil, err + } + accessID, privateKey = readGCSCredentials(credsJSON) + if accessID == "" && metadata.OnGCE() { + accessID, _ = metadata.Email("") + } + } + + g := &GCS{ + bucket: u.Host, + url: urlStr, + client: client, + apiBase: host + "/storage/v1", + uploadBase: host + "/upload/storage/v1", + accessID: accessID, + privateKey: privateKey, + } + if len(privateKey) == 0 && accessID != "" { + g.signBytes = g.signBlob + } + return g, nil +} + +func gcsHTTPClient(ctx context.Context) (*http.Client, []byte, error) { + creds, err := google.FindDefaultCredentials(ctx, gcsScope) + if err != nil { + return nil, nil, fmt.Errorf("loading GCS default credentials: %w", err) + } + return oauth2.NewClient(ctx, creds.TokenSource), creds.JSON, nil +} + +func readGCSCredentials(credFileAsJSON []byte) (string, []byte) { + var serviceAccount struct { + ClientEmail string `json:"client_email"` + PrivateKey string `json:"private_key"` + } + if err := json.Unmarshal(credFileAsJSON, &serviceAccount); err == nil && serviceAccount.ClientEmail != "" { + return serviceAccount.ClientEmail, []byte(serviceAccount.PrivateKey) + } + + var impersonated struct { + ServiceAccountImpersonationURL string `json:"service_account_impersonation_url"` + } + if err := json.Unmarshal(credFileAsJSON, &impersonated); err == nil { + if email := serviceAccountFromImpersonationURL(impersonated.ServiceAccountImpersonationURL); email != "" { + return email, nil + } + } + + return "", nil +} + +func serviceAccountFromImpersonationURL(raw string) string { + if raw == "" { + return "" + } + u, err := url.Parse(raw) + if err != nil { + return "" + } + const marker = "/serviceAccounts/" + idx := strings.Index(u.Path, marker) + if idx == -1 { + return "" + } + email := strings.TrimSuffix(u.Path[idx+len(marker):], ":generateAccessToken") + email, _ = url.PathUnescape(email) + return email +} + +func (g *GCS) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { + h := sha256.New() + body := &countingReader{r: io.TeeReader(r, h)} + + endpoint := g.uploadBase + "/b/" + url.PathEscape(g.bucket) + "/o" + reqURL, _ := url.Parse(endpoint) + q := reqURL.Query() + q.Set("uploadType", "media") + q.Set("name", path) + reqURL.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL.String(), body) + if err != nil { + return 0, "", err + } + req.Header.Set("Content-Type", "application/octet-stream") + + resp, err := g.client.Do(req) + if err != nil { + return 0, "", fmt.Errorf("uploading GCS object: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if err := g.checkResponse(resp, http.StatusOK); err != nil { + return 0, "", fmt.Errorf("uploading GCS object: %w", err) + } + + hash := hex.EncodeToString(h.Sum(nil)) + return body.n, hash, nil +} + +func (g *GCS) Open(ctx context.Context, path string) (io.ReadCloser, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.objectURL(path)+"?alt=media", nil) + if err != nil { + return nil, err + } + resp, err := g.client.Do(req) + if err != nil { + return nil, fmt.Errorf("opening GCS object: %w", err) + } + if resp.StatusCode == http.StatusNotFound { + _ = resp.Body.Close() + return nil, ErrNotFound + } + if err := g.checkResponse(resp, http.StatusOK); err != nil { + _ = resp.Body.Close() + return nil, fmt.Errorf("opening GCS object: %w", err) + } + return resp.Body, nil +} + +func (g *GCS) Exists(ctx context.Context, path string) (bool, error) { + _, err := g.attrs(ctx, path) + if errors.Is(err, ErrNotFound) { + return false, nil + } + return err == nil, err +} + +func (g *GCS) Delete(ctx context.Context, path string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, g.objectURL(path), nil) + if err != nil { + return err + } + resp, err := g.client.Do(req) + if err != nil { + return fmt.Errorf("deleting GCS object: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode == http.StatusNotFound { + return nil + } + if err := g.checkResponse(resp, http.StatusNoContent); err != nil { + return fmt.Errorf("deleting GCS object: %w", err) + } + return nil +} + +func (g *GCS) Size(ctx context.Context, path string) (int64, error) { + obj, err := g.attrs(ctx, path) + if err != nil { + return 0, err + } + return obj.size(), nil +} + +func (g *GCS) SignedURL(ctx context.Context, path string, expiry time.Duration) (string, error) { + if g.accessID == "" { + return "", ErrSignedURLUnsupported + } + + expires := time.Now().Add(expiry) + u := &url.URL{Path: fmt.Sprintf("/%s/%s", g.bucket, path)} + stringToSign := fmt.Sprintf("GET\n\n\n%d\n%s", expires.Unix(), u.String()) + + signed, err := g.sign(ctx, []byte(stringToSign)) + if err != nil { + return "", fmt.Errorf("signing GCS URL: %w", err) + } + + u.Scheme = "https" + u.Host = "storage.googleapis.com" + q := u.Query() + q.Set("GoogleAccessId", g.accessID) + q.Set("Expires", strconv.FormatInt(expires.Unix(), 10)) + q.Set("Signature", base64.StdEncoding.EncodeToString(signed)) + u.RawQuery = q.Encode() + return u.String(), nil +} + +func (g *GCS) UsedSpace(ctx context.Context) (int64, error) { + objects, err := g.ListPrefix(ctx, "") + if err != nil { + return 0, err + } + var total int64 + for _, obj := range objects { + total += obj.Size + } + return total, nil +} + +func (g *GCS) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error) { + var objects []ObjectInfo + pageToken := "" + + for { + reqURL, _ := url.Parse(g.apiBase + "/b/" + url.PathEscape(g.bucket) + "/o") + q := reqURL.Query() + q.Set("prefix", prefix) + if pageToken != "" { + q.Set("pageToken", pageToken) + } + reqURL.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) + if err != nil { + return nil, err + } + resp, err := g.client.Do(req) + if err != nil { + return nil, fmt.Errorf("listing GCS objects: %w", err) + } + if err := g.checkResponse(resp, http.StatusOK); err != nil { + _ = resp.Body.Close() + return nil, fmt.Errorf("listing GCS objects: %w", err) + } + + var page gcsListResponse + if err := json.NewDecoder(resp.Body).Decode(&page); err != nil { + _ = resp.Body.Close() + return nil, fmt.Errorf("decoding GCS list response: %w", err) + } + _ = resp.Body.Close() + + for _, item := range page.Items { + objects = append(objects, ObjectInfo{ + Path: item.Name, + Size: item.size(), + ModTime: item.updated(), + }) + } + if page.NextPageToken == "" { + return objects, nil + } + pageToken = page.NextPageToken + } +} + +func (g *GCS) Close() error { + return nil +} + +func (g *GCS) URL() string { + return g.url +} + +func (g *GCS) attrs(ctx context.Context, path string) (*gcsObject, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.objectURL(path), nil) + if err != nil { + return nil, err + } + resp, err := g.client.Do(req) + if err != nil { + return nil, fmt.Errorf("getting GCS object attributes: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode == http.StatusNotFound { + return nil, ErrNotFound + } + if err := g.checkResponse(resp, http.StatusOK); err != nil { + return nil, fmt.Errorf("getting GCS object attributes: %w", err) + } + + var obj gcsObject + if err := json.NewDecoder(resp.Body).Decode(&obj); err != nil { + return nil, fmt.Errorf("decoding GCS attributes: %w", err) + } + return &obj, nil +} + +func (g *GCS) objectURL(path string) string { + return g.apiBase + "/b/" + url.PathEscape(g.bucket) + "/o/" + url.PathEscape(path) +} + +func (g *GCS) checkResponse(resp *http.Response, want int) error { + if resp.StatusCode == want { + return nil + } + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) +} + +func (g *GCS) sign(ctx context.Context, b []byte) ([]byte, error) { + if len(g.privateKey) > 0 { + key, err := parseGCSPrivateKey(g.privateKey) + if err != nil { + return nil, err + } + sum := sha256.Sum256(b) + return rsa.SignPKCS1v15(rand.Reader, key, crypto.SHA256, sum[:]) + } + if g.signBytes != nil { + return g.signBytes(ctx, b) + } + return nil, ErrSignedURLUnsupported +} + +func (g *GCS) signBlob(ctx context.Context, payload []byte) ([]byte, error) { + reqBody, err := json.Marshal(map[string]string{ + "payload": base64.StdEncoding.EncodeToString(payload), + }) + if err != nil { + return nil, err + } + + endpoint := "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/" + + url.PathEscape(g.accessID) + ":signBlob" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(reqBody)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := g.client.Do(req) + if err != nil { + return nil, fmt.Errorf("calling IAM signBlob: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if err := g.checkResponse(resp, http.StatusOK); err != nil { + return nil, fmt.Errorf("calling IAM signBlob: %w", err) + } + + var out struct { + SignedBlob string `json:"signedBlob"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, fmt.Errorf("decoding IAM signBlob response: %w", err) + } + return base64.StdEncoding.DecodeString(out.SignedBlob) +} + +func parseGCSPrivateKey(key []byte) (*rsa.PrivateKey, error) { + if block, _ := pem.Decode(key); block != nil { + key = block.Bytes + } + parsedKey, err := x509.ParsePKCS8PrivateKey(key) + if err != nil { + parsedKey, err = x509.ParsePKCS1PrivateKey(key) + if err != nil { + return nil, err + } + } + parsed, ok := parsedKey.(*rsa.PrivateKey) + if !ok { + return nil, errors.New("private key is not RSA") + } + return parsed, nil +} + +type gcsObject struct { + Name string `json:"name"` + Size string `json:"size"` + Updated string `json:"updated"` +} + +type countingReader struct { + r io.Reader + n int64 +} + +func (r *countingReader) Read(p []byte) (int, error) { + n, err := r.r.Read(p) + r.n += int64(n) + return n, err +} + +func (o gcsObject) size() int64 { + n, _ := strconv.ParseInt(o.Size, 10, 64) + return n +} + +func (o gcsObject) updated() time.Time { + t, _ := time.Parse(time.RFC3339Nano, o.Updated) + return t +} + +type gcsListResponse struct { + NextPageToken string `json:"nextPageToken"` + Items []gcsObject `json:"items"` +} diff --git a/internal/storage/gcs_test.go b/internal/storage/gcs_test.go new file mode 100644 index 0000000..44dff00 --- /dev/null +++ b/internal/storage/gcs_test.go @@ -0,0 +1,145 @@ +package storage + +import ( + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "sort" + "strconv" + "strings" + "testing" + "time" +) + +func TestGCSRoundTripWithEmulator(t *testing.T) { + objects := map[string]string{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/upload/storage/v1/b/test-bucket/o": + name := r.URL.Query().Get("name") + data, _ := io.ReadAll(r.Body) + objects[name] = string(data) + writeJSON(w, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + case r.Method == http.MethodGet && r.URL.Path == "/storage/v1/b/test-bucket/o": + prefix := r.URL.Query().Get("prefix") + page := gcsListResponse{} + for name, data := range objects { + if strings.HasPrefix(name, prefix) { + page.Items = append(page.Items, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + } + } + sort.Slice(page.Items, func(i, j int) bool { return page.Items[i].Name < page.Items[j].Name }) + writeJSON(w, page) + case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): + name := objectNameFromPath(r.URL.Path) + data, ok := objects[name] + if !ok { + http.NotFound(w, r) + return + } + if r.URL.Query().Get("alt") == "media" { + _, _ = io.WriteString(w, data) + return + } + writeJSON(w, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + case r.Method == http.MethodDelete && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): + delete(objects, objectNameFromPath(r.URL.Path)) + w.WriteHeader(http.StatusNoContent) + default: + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String()) + } + })) + defer server.Close() + t.Setenv("STORAGE_EMULATOR_HOST", server.URL) + + ctx := context.Background() + store, err := OpenGCS(ctx, "gs://test-bucket") + if err != nil { + t.Fatalf("OpenGCS failed: %v", err) + } + + size, hash, err := store.Store(ctx, "npm/pkg/file.tgz", strings.NewReader("content")) + if err != nil { + t.Fatalf("Store failed: %v", err) + } + if size != int64(len("content")) || hash == "" { + t.Fatalf("Store returned size=%d hash=%q", size, hash) + } + + exists, err := store.Exists(ctx, "npm/pkg/file.tgz") + if err != nil || !exists { + t.Fatalf("Exists = %v, %v; want true, nil", exists, err) + } + + r, err := store.Open(ctx, "npm/pkg/file.tgz") + if err != nil { + t.Fatalf("Open failed: %v", err) + } + data, _ := io.ReadAll(r) + _ = r.Close() + if string(data) != "content" { + t.Fatalf("Open content = %q, want content", data) + } + + list, err := store.ListPrefix(ctx, "npm/") + if err != nil { + t.Fatalf("ListPrefix failed: %v", err) + } + if len(list) != 1 || list[0].Path != "npm/pkg/file.tgz" { + t.Fatalf("ListPrefix = %#v", list) + } + + if err := store.Delete(ctx, "npm/pkg/file.tgz"); err != nil { + t.Fatalf("Delete failed: %v", err) + } + exists, err = store.Exists(ctx, "npm/pkg/file.tgz") + if err != nil || exists { + t.Fatalf("Exists after delete = %v, %v; want false, nil", exists, err) + } +} + +func TestGCSSignedURLUsesSigner(t *testing.T) { + store := &GCS{ + bucket: "test-bucket", + accessID: "service@example.com", + signBytes: func(_ context.Context, b []byte) ([]byte, error) { + if !strings.Contains(string(b), "/test-bucket/npm/pkg/file.tgz") { + t.Fatalf("string to sign = %q", b) + } + return []byte("signed"), nil + }, + } + + got, err := store.SignedURL(context.Background(), "npm/pkg/file.tgz", time.Minute) + if err != nil { + t.Fatalf("SignedURL failed: %v", err) + } + u, err := url.Parse(got) + if err != nil { + t.Fatalf("parsing signed URL: %v", err) + } + if u.Scheme != "https" || u.Host != "storage.googleapis.com" || u.Path != "/test-bucket/npm/pkg/file.tgz" { + t.Fatalf("signed URL location = %s", got) + } + if u.Query().Get("GoogleAccessId") != "service@example.com" { + t.Fatalf("GoogleAccessId = %q", u.Query().Get("GoogleAccessId")) + } + if u.Query().Get("Signature") != base64.StdEncoding.EncodeToString([]byte("signed")) { + t.Fatalf("Signature = %q", u.Query().Get("Signature")) + } +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +func objectNameFromPath(p string) string { + escaped := strings.TrimPrefix(p, "/storage/v1/b/test-bucket/o/") + name, _ := url.PathUnescape(escaped) + return name +} From e26c3690db0a62a4be7a7daede2113466af44162 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Fri, 28 Aug 2026 13:38:57 +0100 Subject: [PATCH 3/3] Extract GCS client into standalone module --- README.md | 4 +- go.mod | 5 +- go.sum | 2 + internal/config/config.go | 13 +- internal/server/eviction_test.go | 7 +- internal/storage/blob.go | 61 +---- internal/storage/blob_test.go | 6 +- internal/storage/gcs.go | 426 +++---------------------------- internal/storage/gcs_test.go | 145 ++++++----- 9 files changed, 139 insertions(+), 530 deletions(-) diff --git a/README.md b/README.md index ba70b42..163f3d6 100644 --- a/README.md +++ b/README.md @@ -558,7 +558,7 @@ storage: url: "gs://my-bucket-name" ``` -Authentication uses [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials), which means no credentials need to be embedded in the config or environment. Supported sources, in order: +Authentication uses [Application Default Credentials](https://docs.cloud.google.com/docs/authentication/application-default-credentials), which means no credentials need to be embedded in the config or environment. Supported sources, in order: - **GKE Workload Identity** — bind the Kubernetes service account running the proxy to a Google service account that has `roles/storage.objectAdmin` on the bucket. The proxy will use the workload's token automatically. - **Attached service account** on GCE, Cloud Run, Cloud Functions, etc. @@ -591,7 +591,7 @@ kubectl annotate serviceaccount KSA_NAME \ #### Direct serve (signed URLs) with Workload Identity -When `direct_serve: true` is enabled, the proxy issues HTTP 302 redirects to presigned GCS URLs. Because Workload Identity provides no private key, the gcsblob driver falls back to the [IAM Credentials `signBlob` API](https://cloud.google.com/iam/docs/reference/credentials/rest/v1/projects.serviceAccounts/signBlob). For this to work, grant the service account the token-creator role on itself: +When `direct_serve: true` is enabled, the proxy issues HTTP 302 redirects to presigned GCS URLs. Workload Identity provides no private key, so the GCS backend calls the [IAM Credentials `signBlob` API](https://docs.cloud.google.com/iam/docs/reference/credentials/rest/v1/projects.serviceAccounts/signBlob). Grant the service account the token-creator role on itself: ```bash gcloud iam service-accounts add-iam-policy-binding \ diff --git a/go.mod b/go.mod index c4c740b..0203ed2 100644 --- a/go.mod +++ b/go.mod @@ -1,8 +1,6 @@ module github.com/git-pkgs/proxy -go 1.26.0 - -toolchain go1.26.6 +go 1.26.7 require ( github.com/BurntSushi/toml v1.6.0 @@ -10,6 +8,7 @@ require ( github.com/git-pkgs/archives v0.5.1 github.com/git-pkgs/cooldown v0.2.0 github.com/git-pkgs/enrichment v0.7.0 + github.com/git-pkgs/gcs v0.1.0 github.com/git-pkgs/integrity v0.1.1 github.com/git-pkgs/magic v0.2.0 github.com/git-pkgs/purl v0.1.19 diff --git a/go.sum b/go.sum index 2cdf33b..5e9e2cc 100644 --- a/go.sum +++ b/go.sum @@ -263,6 +263,8 @@ github.com/git-pkgs/cooldown v0.2.0 h1:0MWPHtkzZgvCR0wdiQeyvMea/dxgw9tParH1zzaFo github.com/git-pkgs/cooldown v0.2.0/go.mod h1:v7APuK/UouTiu8mWQZbdDmj7DfxxkGUeuhjaRB5gv9E= github.com/git-pkgs/enrichment v0.7.0 h1:LfIzlVArc2p0MONO08ybC5jiHlysfzS3YyZ5ry2d6Lw= github.com/git-pkgs/enrichment v0.7.0/go.mod h1:ZgZJq7cz1H/nlkgHjCdIRLF/TA5VcbUmpE970VQmG14= +github.com/git-pkgs/gcs v0.1.0 h1:E3awGtsO0xZyHT9FUfEwMHjMkRxw41Bh+c7lgTmPvBo= +github.com/git-pkgs/gcs v0.1.0/go.mod h1:bdkCFD66ryaWnU8MBhokVA3WkJfyAEchm9qec5woBpE= github.com/git-pkgs/integrity v0.1.1 h1:nHQ7SktOiGM1dOb5BFnkdtttG/6FCgE6r5ru6QnsGts= github.com/git-pkgs/integrity v0.1.1/go.mod h1:hxu24lcd230377hCF28JQW7sGcCbuNLqo/0ULeb+F1Q= github.com/git-pkgs/magic v0.2.0 h1:c7HqVxnP8c88EaVMH0/KraDFVTcmiXckRiSvNZEnvMQ= diff --git a/internal/config/config.go b/internal/config/config.go index 89c2616..bedac93 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,13 +33,12 @@ // // AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION // -// For GCS, authentication uses Application Default Credentials. This works -// transparently on GKE with Workload Identity, GCE/Cloud Run with attached -// service accounts, or locally via `gcloud auth application-default login`. -// When the proxy is signing URLs (direct_serve) without a private key, -// gcsblob automatically falls back to the IAM Credentials signBlob API, -// which requires the service account to hold roles/iam.serviceAccountTokenCreator -// on itself. +// For GCS, authentication uses Application Default Credentials. This supports +// GKE Workload Identity, attached service accounts on GCE and Cloud Run, and +// local credentials created by `gcloud auth application-default login`. +// When direct_serve is enabled without a private key, the GCS backend uses the +// IAM Credentials signBlob API. The service account must hold +// roles/iam.serviceAccountTokenCreator on itself. // // Database Configuration: // diff --git a/internal/server/eviction_test.go b/internal/server/eviction_test.go index 80badbe..bac3325 100644 --- a/internal/server/eviction_test.go +++ b/internal/server/eviction_test.go @@ -32,12 +32,17 @@ func setupEvictionTest(t *testing.T) (*database.DB, *storage.Blob) { _ = db.Close() t.Fatalf("failed to create storage: %v", err) } + blob, ok := store.(*storage.Blob) + if !ok { + _ = db.Close() + t.Fatalf("OpenBucket returned %T, want *storage.Blob", store) + } t.Cleanup(func() { _ = db.Close() }) - return db, store + return db, blob } func seedArtifact(t *testing.T, ctx context.Context, db *database.DB, store storage.Storage, name string, dataSize int, accessedAt time.Time) { diff --git a/internal/storage/blob.go b/internal/storage/blob.go index 34f6f08..97e50f3 100644 --- a/internal/storage/blob.go +++ b/internal/storage/blob.go @@ -25,9 +25,8 @@ const osWindows = "windows" // Blob implements Storage using gocloud.dev/blob. // Supports local filesystem (file://) and S3 (s3://) URLs. type Blob struct { - bucket *blob.Bucket - backend Storage - url string + bucket *blob.Bucket + url string } // OpenBucket opens a blob bucket from a URL. @@ -41,13 +40,11 @@ type Blob struct { // - azblob://container-name - Azure Blob Storage // // For local filesystem, the directory is created if it doesn't exist. -func OpenBucket(ctx context.Context, urlStr string) (*Blob, error) { +// +//nolint:ireturn // The URL scheme selects the storage implementation. +func OpenBucket(ctx context.Context, urlStr string) (Storage, error) { if strings.HasPrefix(urlStr, "gs://") { - backend, err := OpenGCS(ctx, urlStr) - if err != nil { - return nil, err - } - return &Blob{backend: backend, url: urlStr}, nil + return OpenGCS(ctx, urlStr) } // Handle file:// URLs specially to create the directory @@ -102,10 +99,6 @@ func OpenBucket(ctx context.Context, urlStr string) (*Blob, error) { } func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { - if b.backend != nil { - return b.backend.Store(ctx, path, r) - } - // Compute hash while writing h := sha256.New() tee := io.TeeReader(r, h) @@ -131,10 +124,6 @@ func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, stri } func (b *Blob) Open(ctx context.Context, path string) (io.ReadCloser, error) { - if b.backend != nil { - return b.backend.Open(ctx, path) - } - r, err := b.bucket.NewReader(ctx, path, nil) if err != nil { if isNotExist(err) { @@ -146,10 +135,6 @@ func (b *Blob) Open(ctx context.Context, path string) (io.ReadCloser, error) { } func (b *Blob) Exists(ctx context.Context, path string) (bool, error) { - if b.backend != nil { - return b.backend.Exists(ctx, path) - } - exists, err := b.bucket.Exists(ctx, path) if err != nil { return false, fmt.Errorf("checking existence: %w", err) @@ -158,10 +143,6 @@ func (b *Blob) Exists(ctx context.Context, path string) (bool, error) { } func (b *Blob) Delete(ctx context.Context, path string) error { - if b.backend != nil { - return b.backend.Delete(ctx, path) - } - err := b.bucket.Delete(ctx, path) if err != nil && !isNotExist(err) { return fmt.Errorf("deleting object: %w", err) @@ -170,10 +151,6 @@ func (b *Blob) Delete(ctx context.Context, path string) error { } func (b *Blob) SignedURL(ctx context.Context, path string, expiry time.Duration) (string, error) { - if b.backend != nil { - return b.backend.SignedURL(ctx, path, expiry) - } - url, err := b.bucket.SignedURL(ctx, path, &blob.SignedURLOptions{ Method: http.MethodGet, Expiry: expiry, @@ -188,10 +165,6 @@ func (b *Blob) SignedURL(ctx context.Context, path string, expiry time.Duration) } func (b *Blob) Size(ctx context.Context, path string) (int64, error) { - if b.backend != nil { - return b.backend.Size(ctx, path) - } - attrs, err := b.bucket.Attributes(ctx, path) if err != nil { if isNotExist(err) { @@ -203,10 +176,6 @@ func (b *Blob) Size(ctx context.Context, path string) (int64, error) { } func (b *Blob) UsedSpace(ctx context.Context) (int64, error) { - if b.backend != nil { - return b.backend.UsedSpace(ctx) - } - var total int64 iter := b.bucket.List(nil) @@ -226,16 +195,6 @@ func (b *Blob) UsedSpace(ctx context.Context) (int64, error) { // ListPrefix returns object metadata for keys under a prefix. func (b *Blob) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error) { - if b.backend != nil { - lister, ok := b.backend.(interface { - ListPrefix(context.Context, string) ([]ObjectInfo, error) - }) - if !ok { - return nil, ErrNotFound - } - return lister.ListPrefix(ctx, prefix) - } - iter := b.bucket.List(&blob.ListOptions{Prefix: prefix}) objects := make([]ObjectInfo, 0) @@ -264,18 +223,10 @@ func (b *Blob) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, err } func (b *Blob) Close() error { - if b.backend != nil { - return b.backend.Close() - } - return b.bucket.Close() } func (b *Blob) URL() string { - if b.backend != nil { - return b.backend.URL() - } - return b.url } diff --git a/internal/storage/blob_test.go b/internal/storage/blob_test.go index d80290b..3e5bf65 100644 --- a/internal/storage/blob_test.go +++ b/internal/storage/blob_test.go @@ -278,7 +278,11 @@ func createTestBlob(t *testing.T) *Blob { t.Fatalf("OpenBucket failed: %v", err) } t.Cleanup(func() { _ = b.Close() }) - return b + blob, ok := b.(*Blob) + if !ok { + t.Fatalf("OpenBucket returned %T, want *Blob", b) + } + return blob } func fileURLFromPath(path string) string { diff --git a/internal/storage/gcs.go b/internal/storage/gcs.go index e97e772..a096c18 100644 --- a/internal/storage/gcs.go +++ b/internal/storage/gcs.go @@ -1,314 +1,91 @@ package storage import ( - "bytes" "context" - "crypto" - "crypto/rand" - "crypto/rsa" "crypto/sha256" - "crypto/x509" - "encoding/base64" "encoding/hex" - "encoding/json" - "encoding/pem" "errors" - "fmt" "io" - "net/http" - "net/url" - "os" - "strconv" - "strings" "time" - "cloud.google.com/go/compute/metadata" - "golang.org/x/oauth2" - "golang.org/x/oauth2/google" + gcstorage "github.com/git-pkgs/gcs" ) -const ( - gcsScope = "https://www.googleapis.com/auth/cloud-platform" - gcsDefaultHost = "https://storage.googleapis.com" -) - -// GCS implements Storage using the Cloud Storage JSON API. +// GCS adapts a Google Cloud Storage bucket to Storage. type GCS struct { - bucket string - url string - client *http.Client - apiBase string - uploadBase string - - accessID string - privateKey []byte - signBytes func(context.Context, []byte) ([]byte, error) + bucket *gcstorage.Bucket + url string } // OpenGCS opens a Google Cloud Storage bucket from a gs:// URL. func OpenGCS(ctx context.Context, urlStr string) (*GCS, error) { - u, err := url.Parse(urlStr) - if err != nil { - return nil, fmt.Errorf("parsing GCS URL: %w", err) - } - if u.Scheme != "gs" || u.Host == "" { - return nil, fmt.Errorf("invalid GCS URL %q", urlStr) - } - if u.Path != "" && u.Path != "/" { - return nil, fmt.Errorf("GCS URL must name a bucket, got path %q", u.Path) - } - - host := strings.TrimRight(gcsDefaultHost, "/") - client := http.DefaultClient - var accessID string - var privateKey []byte - - if emulator := os.Getenv("STORAGE_EMULATOR_HOST"); emulator != "" { - host = strings.TrimRight(emulator, "/") - if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") { - host = "http://" + host - } - } else { - var credsJSON []byte - var err error - client, credsJSON, err = gcsHTTPClient(ctx) - if err != nil { - return nil, err - } - accessID, privateKey = readGCSCredentials(credsJSON) - if accessID == "" && metadata.OnGCE() { - accessID, _ = metadata.Email("") - } - } - - g := &GCS{ - bucket: u.Host, - url: urlStr, - client: client, - apiBase: host + "/storage/v1", - uploadBase: host + "/upload/storage/v1", - accessID: accessID, - privateKey: privateKey, - } - if len(privateKey) == 0 && accessID != "" { - g.signBytes = g.signBlob - } - return g, nil -} - -func gcsHTTPClient(ctx context.Context) (*http.Client, []byte, error) { - creds, err := google.FindDefaultCredentials(ctx, gcsScope) + bucket, err := gcstorage.OpenBucket(ctx, urlStr) if err != nil { - return nil, nil, fmt.Errorf("loading GCS default credentials: %w", err) - } - return oauth2.NewClient(ctx, creds.TokenSource), creds.JSON, nil -} - -func readGCSCredentials(credFileAsJSON []byte) (string, []byte) { - var serviceAccount struct { - ClientEmail string `json:"client_email"` - PrivateKey string `json:"private_key"` - } - if err := json.Unmarshal(credFileAsJSON, &serviceAccount); err == nil && serviceAccount.ClientEmail != "" { - return serviceAccount.ClientEmail, []byte(serviceAccount.PrivateKey) - } - - var impersonated struct { - ServiceAccountImpersonationURL string `json:"service_account_impersonation_url"` - } - if err := json.Unmarshal(credFileAsJSON, &impersonated); err == nil { - if email := serviceAccountFromImpersonationURL(impersonated.ServiceAccountImpersonationURL); email != "" { - return email, nil - } - } - - return "", nil -} - -func serviceAccountFromImpersonationURL(raw string) string { - if raw == "" { - return "" - } - u, err := url.Parse(raw) - if err != nil { - return "" - } - const marker = "/serviceAccounts/" - idx := strings.Index(u.Path, marker) - if idx == -1 { - return "" + return nil, err } - email := strings.TrimSuffix(u.Path[idx+len(marker):], ":generateAccessToken") - email, _ = url.PathUnescape(email) - return email + return &GCS{bucket: bucket, url: urlStr}, nil } func (g *GCS) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { h := sha256.New() - body := &countingReader{r: io.TeeReader(r, h)} - - endpoint := g.uploadBase + "/b/" + url.PathEscape(g.bucket) + "/o" - reqURL, _ := url.Parse(endpoint) - q := reqURL.Query() - q.Set("uploadType", "media") - q.Set("name", path) - reqURL.RawQuery = q.Encode() - - req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL.String(), body) + size, err := g.bucket.Write(ctx, path, io.TeeReader(r, h)) if err != nil { return 0, "", err } - req.Header.Set("Content-Type", "application/octet-stream") - - resp, err := g.client.Do(req) - if err != nil { - return 0, "", fmt.Errorf("uploading GCS object: %w", err) - } - defer func() { _ = resp.Body.Close() }() - if err := g.checkResponse(resp, http.StatusOK); err != nil { - return 0, "", fmt.Errorf("uploading GCS object: %w", err) - } - - hash := hex.EncodeToString(h.Sum(nil)) - return body.n, hash, nil + return size, hex.EncodeToString(h.Sum(nil)), nil } func (g *GCS) Open(ctx context.Context, path string) (io.ReadCloser, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.objectURL(path)+"?alt=media", nil) - if err != nil { - return nil, err - } - resp, err := g.client.Do(req) - if err != nil { - return nil, fmt.Errorf("opening GCS object: %w", err) - } - if resp.StatusCode == http.StatusNotFound { - _ = resp.Body.Close() + r, err := g.bucket.Open(ctx, path) + if errors.Is(err, gcstorage.ErrNotFound) { return nil, ErrNotFound } - if err := g.checkResponse(resp, http.StatusOK); err != nil { - _ = resp.Body.Close() - return nil, fmt.Errorf("opening GCS object: %w", err) - } - return resp.Body, nil + return r, err } func (g *GCS) Exists(ctx context.Context, path string) (bool, error) { - _, err := g.attrs(ctx, path) - if errors.Is(err, ErrNotFound) { - return false, nil - } - return err == nil, err + return g.bucket.Exists(ctx, path) } func (g *GCS) Delete(ctx context.Context, path string) error { - req, err := http.NewRequestWithContext(ctx, http.MethodDelete, g.objectURL(path), nil) - if err != nil { - return err - } - resp, err := g.client.Do(req) - if err != nil { - return fmt.Errorf("deleting GCS object: %w", err) - } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode == http.StatusNotFound { - return nil - } - if err := g.checkResponse(resp, http.StatusNoContent); err != nil { - return fmt.Errorf("deleting GCS object: %w", err) - } - return nil + return g.bucket.Delete(ctx, path) } func (g *GCS) Size(ctx context.Context, path string) (int64, error) { - obj, err := g.attrs(ctx, path) - if err != nil { - return 0, err + size, err := g.bucket.Size(ctx, path) + if errors.Is(err, gcstorage.ErrNotFound) { + return 0, ErrNotFound } - return obj.size(), nil + return size, err } func (g *GCS) SignedURL(ctx context.Context, path string, expiry time.Duration) (string, error) { - if g.accessID == "" { + u, err := g.bucket.SignedURL(ctx, path, expiry) + if errors.Is(err, gcstorage.ErrSignedURLUnsupported) { return "", ErrSignedURLUnsupported } - - expires := time.Now().Add(expiry) - u := &url.URL{Path: fmt.Sprintf("/%s/%s", g.bucket, path)} - stringToSign := fmt.Sprintf("GET\n\n\n%d\n%s", expires.Unix(), u.String()) - - signed, err := g.sign(ctx, []byte(stringToSign)) - if err != nil { - return "", fmt.Errorf("signing GCS URL: %w", err) - } - - u.Scheme = "https" - u.Host = "storage.googleapis.com" - q := u.Query() - q.Set("GoogleAccessId", g.accessID) - q.Set("Expires", strconv.FormatInt(expires.Unix(), 10)) - q.Set("Signature", base64.StdEncoding.EncodeToString(signed)) - u.RawQuery = q.Encode() - return u.String(), nil + return u, err } func (g *GCS) UsedSpace(ctx context.Context) (int64, error) { - objects, err := g.ListPrefix(ctx, "") - if err != nil { - return 0, err - } - var total int64 - for _, obj := range objects { - total += obj.Size - } - return total, nil + return g.bucket.UsedSpace(ctx) } func (g *GCS) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error) { - var objects []ObjectInfo - pageToken := "" - - for { - reqURL, _ := url.Parse(g.apiBase + "/b/" + url.PathEscape(g.bucket) + "/o") - q := reqURL.Query() - q.Set("prefix", prefix) - if pageToken != "" { - q.Set("pageToken", pageToken) - } - reqURL.RawQuery = q.Encode() - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) - if err != nil { - return nil, err - } - resp, err := g.client.Do(req) - if err != nil { - return nil, fmt.Errorf("listing GCS objects: %w", err) - } - if err := g.checkResponse(resp, http.StatusOK); err != nil { - _ = resp.Body.Close() - return nil, fmt.Errorf("listing GCS objects: %w", err) - } - - var page gcsListResponse - if err := json.NewDecoder(resp.Body).Decode(&page); err != nil { - _ = resp.Body.Close() - return nil, fmt.Errorf("decoding GCS list response: %w", err) - } - _ = resp.Body.Close() + objects, err := g.bucket.ListPrefix(ctx, prefix) + if err != nil { + return nil, err + } - for _, item := range page.Items { - objects = append(objects, ObjectInfo{ - Path: item.Name, - Size: item.size(), - ModTime: item.updated(), - }) - } - if page.NextPageToken == "" { - return objects, nil - } - pageToken = page.NextPageToken + result := make([]ObjectInfo, 0, len(objects)) + for _, object := range objects { + result = append(result, ObjectInfo{ + Path: object.Name, + Size: object.Size, + ModTime: object.ModTime, + }) } + return result, nil } func (g *GCS) Close() error { @@ -318,138 +95,3 @@ func (g *GCS) Close() error { func (g *GCS) URL() string { return g.url } - -func (g *GCS) attrs(ctx context.Context, path string) (*gcsObject, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.objectURL(path), nil) - if err != nil { - return nil, err - } - resp, err := g.client.Do(req) - if err != nil { - return nil, fmt.Errorf("getting GCS object attributes: %w", err) - } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode == http.StatusNotFound { - return nil, ErrNotFound - } - if err := g.checkResponse(resp, http.StatusOK); err != nil { - return nil, fmt.Errorf("getting GCS object attributes: %w", err) - } - - var obj gcsObject - if err := json.NewDecoder(resp.Body).Decode(&obj); err != nil { - return nil, fmt.Errorf("decoding GCS attributes: %w", err) - } - return &obj, nil -} - -func (g *GCS) objectURL(path string) string { - return g.apiBase + "/b/" + url.PathEscape(g.bucket) + "/o/" + url.PathEscape(path) -} - -func (g *GCS) checkResponse(resp *http.Response, want int) error { - if resp.StatusCode == want { - return nil - } - body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) -} - -func (g *GCS) sign(ctx context.Context, b []byte) ([]byte, error) { - if len(g.privateKey) > 0 { - key, err := parseGCSPrivateKey(g.privateKey) - if err != nil { - return nil, err - } - sum := sha256.Sum256(b) - return rsa.SignPKCS1v15(rand.Reader, key, crypto.SHA256, sum[:]) - } - if g.signBytes != nil { - return g.signBytes(ctx, b) - } - return nil, ErrSignedURLUnsupported -} - -func (g *GCS) signBlob(ctx context.Context, payload []byte) ([]byte, error) { - reqBody, err := json.Marshal(map[string]string{ - "payload": base64.StdEncoding.EncodeToString(payload), - }) - if err != nil { - return nil, err - } - - endpoint := "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/" + - url.PathEscape(g.accessID) + ":signBlob" - req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(reqBody)) - if err != nil { - return nil, err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := g.client.Do(req) - if err != nil { - return nil, fmt.Errorf("calling IAM signBlob: %w", err) - } - defer func() { _ = resp.Body.Close() }() - if err := g.checkResponse(resp, http.StatusOK); err != nil { - return nil, fmt.Errorf("calling IAM signBlob: %w", err) - } - - var out struct { - SignedBlob string `json:"signedBlob"` - } - if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { - return nil, fmt.Errorf("decoding IAM signBlob response: %w", err) - } - return base64.StdEncoding.DecodeString(out.SignedBlob) -} - -func parseGCSPrivateKey(key []byte) (*rsa.PrivateKey, error) { - if block, _ := pem.Decode(key); block != nil { - key = block.Bytes - } - parsedKey, err := x509.ParsePKCS8PrivateKey(key) - if err != nil { - parsedKey, err = x509.ParsePKCS1PrivateKey(key) - if err != nil { - return nil, err - } - } - parsed, ok := parsedKey.(*rsa.PrivateKey) - if !ok { - return nil, errors.New("private key is not RSA") - } - return parsed, nil -} - -type gcsObject struct { - Name string `json:"name"` - Size string `json:"size"` - Updated string `json:"updated"` -} - -type countingReader struct { - r io.Reader - n int64 -} - -func (r *countingReader) Read(p []byte) (int, error) { - n, err := r.r.Read(p) - r.n += int64(n) - return n, err -} - -func (o gcsObject) size() int64 { - n, _ := strconv.ParseInt(o.Size, 10, 64) - return n -} - -func (o gcsObject) updated() time.Time { - t, _ := time.Parse(time.RFC3339Nano, o.Updated) - return t -} - -type gcsListResponse struct { - NextPageToken string `json:"nextPageToken"` - Items []gcsObject `json:"items"` -} diff --git a/internal/storage/gcs_test.go b/internal/storage/gcs_test.go index 44dff00..1dfd0a2 100644 --- a/internal/storage/gcs_test.go +++ b/internal/storage/gcs_test.go @@ -2,8 +2,10 @@ package storage import ( "context" - "encoding/base64" + "crypto/sha256" + "encoding/hex" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -15,58 +17,23 @@ import ( "time" ) -func TestGCSRoundTripWithEmulator(t *testing.T) { - objects := map[string]string{} - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - switch { - case r.Method == http.MethodPost && r.URL.Path == "/upload/storage/v1/b/test-bucket/o": - name := r.URL.Query().Get("name") - data, _ := io.ReadAll(r.Body) - objects[name] = string(data) - writeJSON(w, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) - case r.Method == http.MethodGet && r.URL.Path == "/storage/v1/b/test-bucket/o": - prefix := r.URL.Query().Get("prefix") - page := gcsListResponse{} - for name, data := range objects { - if strings.HasPrefix(name, prefix) { - page.Items = append(page.Items, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) - } - } - sort.Slice(page.Items, func(i, j int) bool { return page.Items[i].Name < page.Items[j].Name }) - writeJSON(w, page) - case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): - name := objectNameFromPath(r.URL.Path) - data, ok := objects[name] - if !ok { - http.NotFound(w, r) - return - } - if r.URL.Query().Get("alt") == "media" { - _, _ = io.WriteString(w, data) - return - } - writeJSON(w, gcsObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) - case r.Method == http.MethodDelete && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): - delete(objects, objectNameFromPath(r.URL.Path)) - w.WriteHeader(http.StatusNoContent) - default: - t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String()) - } - })) +func TestOpenBucketGCSRoundTripWithEmulator(t *testing.T) { + server := httptest.NewServer(&fakeGCSServer{t: t, objects: map[string]string{}}) defer server.Close() t.Setenv("STORAGE_EMULATOR_HOST", server.URL) ctx := context.Background() - store, err := OpenGCS(ctx, "gs://test-bucket") + store, err := OpenBucket(ctx, "gs://test-bucket") if err != nil { - t.Fatalf("OpenGCS failed: %v", err) + t.Fatalf("OpenBucket failed: %v", err) } size, hash, err := store.Store(ctx, "npm/pkg/file.tgz", strings.NewReader("content")) if err != nil { t.Fatalf("Store failed: %v", err) } - if size != int64(len("content")) || hash == "" { + wantHash := sha256.Sum256([]byte("content")) + if size != int64(len("content")) || hash != hex.EncodeToString(wantHash[:]) { t.Fatalf("Store returned size=%d hash=%q", size, hash) } @@ -85,7 +52,13 @@ func TestGCSRoundTripWithEmulator(t *testing.T) { t.Fatalf("Open content = %q, want content", data) } - list, err := store.ListPrefix(ctx, "npm/") + lister, ok := store.(interface { + ListPrefix(context.Context, string) ([]ObjectInfo, error) + }) + if !ok { + t.Fatal("GCS storage does not support prefix listing") + } + list, err := lister.ListPrefix(ctx, "npm/") if err != nil { t.Fatalf("ListPrefix failed: %v", err) } @@ -100,39 +73,73 @@ func TestGCSRoundTripWithEmulator(t *testing.T) { if err != nil || exists { t.Fatalf("Exists after delete = %v, %v; want false, nil", exists, err) } -} - -func TestGCSSignedURLUsesSigner(t *testing.T) { - store := &GCS{ - bucket: "test-bucket", - accessID: "service@example.com", - signBytes: func(_ context.Context, b []byte) ([]byte, error) { - if !strings.Contains(string(b), "/test-bucket/npm/pkg/file.tgz") { - t.Fatalf("string to sign = %q", b) - } - return []byte("signed"), nil - }, - } - got, err := store.SignedURL(context.Background(), "npm/pkg/file.tgz", time.Minute) - if err != nil { - t.Fatalf("SignedURL failed: %v", err) - } - u, err := url.Parse(got) - if err != nil { - t.Fatalf("parsing signed URL: %v", err) + reader, err := store.Open(ctx, "npm/pkg/file.tgz") + if reader != nil || !errors.Is(err, ErrNotFound) { + t.Fatalf("Open missing object = %v, %v; want nil, ErrNotFound", reader, err) } - if u.Scheme != "https" || u.Host != "storage.googleapis.com" || u.Path != "/test-bucket/npm/pkg/file.tgz" { - t.Fatalf("signed URL location = %s", got) + if _, err := store.Size(ctx, "npm/pkg/file.tgz"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Size missing object = %v, want ErrNotFound", err) } - if u.Query().Get("GoogleAccessId") != "service@example.com" { - t.Fatalf("GoogleAccessId = %q", u.Query().Get("GoogleAccessId")) + if _, err := store.SignedURL(ctx, "npm/pkg/file.tgz", time.Minute); !errors.Is(err, ErrSignedURLUnsupported) { + t.Fatalf("SignedURL with emulator = %v, want ErrSignedURLUnsupported", err) } - if u.Query().Get("Signature") != base64.StdEncoding.EncodeToString([]byte("signed")) { - t.Fatalf("Signature = %q", u.Query().Get("Signature")) +} + +type fakeGCSServer struct { + t *testing.T + objects map[string]string +} + +func (f *fakeGCSServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/upload/storage/v1/b/test-bucket/o": + name := r.URL.Query().Get("name") + data, _ := io.ReadAll(r.Body) + f.objects[name] = string(data) + writeJSON(w, fakeGCSObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + case r.Method == http.MethodGet && r.URL.Path == "/storage/v1/b/test-bucket/o": + prefix := r.URL.Query().Get("prefix") + page := fakeGCSListResponse{} + for name, data := range f.objects { + if strings.HasPrefix(name, prefix) { + page.Items = append(page.Items, fakeGCSObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + } + } + sort.Slice(page.Items, func(i, j int) bool { return page.Items[i].Name < page.Items[j].Name }) + writeJSON(w, page) + case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): + name := objectNameFromPath(r.URL.Path) + data, ok := f.objects[name] + if !ok { + http.NotFound(w, r) + return + } + if r.URL.Query().Get("alt") == "media" { + _, _ = io.WriteString(w, data) + return + } + writeJSON(w, fakeGCSObject{Name: name, Size: strconv.Itoa(len(data)), Updated: time.Now().UTC().Format(time.RFC3339Nano)}) + case r.Method == http.MethodDelete && strings.HasPrefix(r.URL.Path, "/storage/v1/b/test-bucket/o/"): + delete(f.objects, objectNameFromPath(r.URL.Path)) + w.WriteHeader(http.StatusNoContent) + default: + f.t.Errorf("unexpected request: %s %s", r.Method, r.URL.String()) + http.Error(w, "unexpected request", http.StatusInternalServerError) } } +type fakeGCSObject struct { + Name string `json:"name"` + Size string `json:"size"` + Updated string `json:"updated"` +} + +type fakeGCSListResponse struct { + NextPageToken string `json:"nextPageToken"` + Items []fakeGCSObject `json:"items"` +} + func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(v)