From e3ae763ea0dd072d06f796e52cf6c920217ae6ad Mon Sep 17 00:00:00 2001 From: erezrokah Date: Fri, 21 Aug 2026 20:15:41 +0100 Subject: [PATCH 1/2] fix: Retry plugin downloads that drop mid-body An HTTP/2 INTERNAL_ERROR stream reset during the body copy was classified as permanent, so the download failed after a single attempt. Classify transient transport failures and retryable status codes for retry, keep 4xx permanent, reset the output file between attempts, verify the body length against Content-Length, and extract the plugin binary through a temporary file so a partial extraction is never cached as a plugin. --- managedplugin/download.go | 118 +++++++++-------- managedplugin/download_retry.go | 109 ++++++++++++++++ managedplugin/download_retry_test.go | 184 +++++++++++++++++++++++++++ 3 files changed, 358 insertions(+), 53 deletions(-) create mode 100644 managedplugin/download_retry.go create mode 100644 managedplugin/download_retry_test.go diff --git a/managedplugin/download.go b/managedplugin/download.go index 3ca05b6..e989285 100644 --- a/managedplugin/download.go +++ b/managedplugin/download.go @@ -8,7 +8,6 @@ import ( "fmt" "io" "net/http" - "net/url" "os" "path/filepath" "runtime" @@ -25,6 +24,7 @@ const ( DefaultDownloadDir = ".cq" RetryAttempts = 5 RetryWaitTime = 1 * time.Second + MaxRetryWaitTime = 8 * time.Second ) func APIBaseURL() string { @@ -190,6 +190,8 @@ func doDownloadPluginFromHub(ctx context.Context, logger zerolog.Logger, c *clou return errors.New("failed to get plugin metadata from hub: empty location from response") } pluginZipPath := ops.LocalPath + ".zip" + defer os.Remove(pluginZipPath) + writtenChecksum, err := downloadFile(ctx, pluginZipPath, location, dops) if err != nil { return fmt.Errorf("failed to download plugin: %w", err) @@ -201,29 +203,46 @@ func doDownloadPluginFromHub(ctx context.Context, logger zerolog.Logger, c *clou return fmt.Errorf("checksum mismatch: expected %s, got %s", pluginAsset.Checksum, writtenChecksum) } - archive, err := zip.OpenReader(pluginZipPath) + pathInArchive := fmt.Sprintf("plugin-%s-%s-%s-%s", ops.PluginName, ops.PluginVersion, runtime.GOOS, runtime.GOARCH) + return extractPluginBinary(pluginZipPath, pathInArchive, ops.LocalPath) +} + +// extractPluginBinary writes the binary to a temporary file and renames it into +// place, so a failure part way through never leaves a truncated binary that the +// next run treats as a cached plugin. +func extractPluginBinary(archivePath, pathInArchive, localPath string) error { + archive, err := zip.OpenReader(archivePath) if err != nil { return fmt.Errorf("failed to open plugin archive: %w", err) } defer archive.Close() - fileInArchive, err := archive.Open(fmt.Sprintf("plugin-%s-%s-%s-%s", ops.PluginName, ops.PluginVersion, runtime.GOOS, runtime.GOARCH)) + fileInArchive, err := archive.Open(pathInArchive) if err != nil { - return fmt.Errorf("failed to open plugin archive: %w", err) + return fmt.Errorf("failed to open plugin archive %s: %w", pathInArchive, err) } + defer fileInArchive.Close() - out, err := os.OpenFile(ops.LocalPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0744) + out, err := os.CreateTemp(filepath.Dir(localPath), filepath.Base(localPath)+".tmp") if err != nil { - return fmt.Errorf("failed to create file %s: %w", ops.LocalPath, err) + return fmt.Errorf("failed to create file %s: %w", localPath, err) } - _, err = io.Copy(out, fileInArchive) - if err != nil { + tmpPath := out.Name() + defer os.Remove(tmpPath) + + if _, err := io.Copy(out, fileInArchive); err != nil { + out.Close() return fmt.Errorf("failed to copy body to file: %w", err) } - err = out.Close() - if err != nil { + if err := out.Close(); err != nil { return fmt.Errorf("failed to close file: %w", err) } + if err := os.Chmod(tmpPath, 0744); err != nil { + return fmt.Errorf("failed to set permissions on %s: %w", localPath, err) + } + if err := os.Rename(tmpPath, localPath); err != nil { + return fmt.Errorf("failed to move plugin binary to %s: %w", localPath, err) + } return nil } @@ -283,17 +302,13 @@ func doDownloadPluginFromGithub(ctx context.Context, logger zerolog.Logger, loca if err != nil { return fmt.Errorf("failed to get plugin url: %w", err) } - logger.Debug().Msg(fmt.Sprintf("Downloading %s", downloadURL)) + logger.Debug().Msg(fmt.Sprintf("Downloading %s", redactURLQuery(downloadURL))) + defer os.Remove(pluginZipPath) + if _, err := downloadFile(ctx, pluginZipPath, downloadURL, dops); err != nil { return fmt.Errorf("failed to download plugin: %w", err) } - archive, err := zip.OpenReader(pluginZipPath) - if err != nil { - return fmt.Errorf("failed to open plugin archive: %w", err) - } - defer archive.Close() - var pathInArchive string switch { case strings.HasPrefix(downloadURL, "https://github.com/cloudquery/cloudquery/releases/download/plugins-plugin"): @@ -312,24 +327,7 @@ func doDownloadPluginFromGithub(ctx context.Context, logger zerolog.Logger, loca return fmt.Errorf("unknown GitHub %s", downloadURL) } - pathInArchive = WithBinarySuffix(pathInArchive) - fileInArchive, err := archive.Open(pathInArchive) - if err != nil { - return fmt.Errorf("failed to open plugin archive plugins/source/%s: %w", name, err) - } - out, err := os.OpenFile(localPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0744) - if err != nil { - return fmt.Errorf("failed to create file %s: %w", localPath, err) - } - _, err = io.Copy(out, fileInArchive) - if err != nil { - return fmt.Errorf("failed to copy body to file: %w", err) - } - err = out.Close() - if err != nil { - return fmt.Errorf("failed to close file: %w", err) - } - return nil + return extractPluginBinary(pluginZipPath, WithBinarySuffix(pathInArchive), localPath) } func downloadFile(ctx context.Context, localPath string, downloadURL string, dops DownloaderOptions) (string, error) { @@ -340,48 +338,49 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop } defer out.Close() - errStatusCodeNotOK := errors.New("statusCode != 200") - errNotFound := errors.New("not found") + urlForLog := redactURLQuery(downloadURL) checksum := "" options := []retry.Option{ - retry.RetryIf(func(err error) bool { - return errors.Is(err, errStatusCodeNotOK) - }), + retry.RetryIf(isRetryableDownloadError), retry.Context(ctx), retry.Attempts(RetryAttempts), retry.Delay(RetryWaitTime), + retry.MaxDelay(MaxRetryWaitTime), } retrier := retry.New(options...) err = retrier.Do(func() error { checksum = "" + // Each attempt rewrites the file from the start, so a body that was cut off + // mid-copy cannot leave its bytes in front of the next attempt's download. + if err := truncateFile(out); err != nil { + return err + } + // Get the data req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil) if err != nil { - return fmt.Errorf("failed create request %s: %w", downloadURL, err) + return fmt.Errorf("failed create request %s: %w", urlForLog, err) } // Do http request resp, err := http.DefaultClient.Do(req) if err != nil { - return fmt.Errorf("failed to get url %s: %w", downloadURL, err) + return fmt.Errorf("failed to get url %s: %w", urlForLog, err) } defer resp.Body.Close() // Check server response if resp.StatusCode == http.StatusNotFound { return errNotFound } else if resp.StatusCode != http.StatusOK { - fmt.Printf("Failed downloading %s with status code %d. Retrying\n", downloadURL, resp.StatusCode) - return errStatusCodeNotOK + if isRetryableStatusCode(resp.StatusCode) { + fmt.Printf("Failed downloading %s with status code %d. Retrying\n", urlForLog, resp.StatusCode) + } else { + fmt.Printf("Failed downloading %s with status code %d\n", urlForLog, resp.StatusCode) + } + return &httpStatusError{statusCode: resp.StatusCode} } - urlForLog := downloadURL - parsedURL, err := url.Parse(downloadURL) - if err == nil { - parsedURL.RawQuery = "" - parsedURL.Fragment = "" - urlForLog = parsedURL.String() - } fmt.Printf("Downloading %s\n", urlForLog) s := sha256.New() @@ -393,10 +392,13 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop } // Write the body to file - _, err = io.Copy(io.MultiWriter(writers...), resp.Body) + written, err := io.Copy(io.MultiWriter(writers...), resp.Body) if err != nil { return fmt.Errorf("failed to copy body to file %s: %w", out.Name(), err) } + if resp.ContentLength >= 0 && written != resp.ContentLength { + return fmt.Errorf("%w: %s got %d bytes, want %d", errShortRead, out.Name(), written, resp.ContentLength) + } checksum = fmt.Sprintf("%x", s.Sum(nil)) return nil }) @@ -404,11 +406,21 @@ func downloadFile(ctx context.Context, localPath string, downloadURL string, dop if errors.Is(err, errNotFound) { return "", errNotFound } - return "", fmt.Errorf("failed downloading URL %q. Error %w", downloadURL, err) + return "", fmt.Errorf("failed downloading URL %q. Error %w", urlForLog, err) } return checksum, nil } +func truncateFile(f *os.File) error { + if err := f.Truncate(0); err != nil { + return fmt.Errorf("failed to truncate file %s: %w", f.Name(), err) + } + if _, err := f.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("failed to rewind file %s: %w", f.Name(), err) + } + return nil +} + func downloadProgressBar(maxBytes int64, description ...string) *progressbar.ProgressBar { desc := "" if len(description) > 0 { diff --git a/managedplugin/download_retry.go b/managedplugin/download_retry.go new file mode 100644 index 0000000..6a9969e --- /dev/null +++ b/managedplugin/download_retry.go @@ -0,0 +1,109 @@ +package managedplugin + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "syscall" +) + +var ( + errNotFound = errors.New("not found") + errShortRead = errors.New("truncated response body") +) + +type httpStatusError struct { + statusCode int +} + +func (e *httpStatusError) Error() string { + return fmt.Sprintf("statusCode %d", e.statusCode) +} + +func isRetryableStatusCode(statusCode int) bool { + switch statusCode { + case http.StatusRequestTimeout, http.StatusTooManyRequests: + return true + } + return statusCode >= http.StatusInternalServerError +} + +// Go does not export the HTTP/2 stream and connection error types used by the +// net/http transport, so the mid-body resets we get from the asset CDN can only +// be matched on their message. +var transientTransportMessages = []string{ + "stream error", + "server sent goaway", + "connection reset by peer", + "broken pipe", + "unexpected eof", + "use of closed network connection", + "server closed idle connection", + "transport connection broken", + "i/o timeout", +} + +func isRetryableDownloadError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + if errors.Is(err, errNotFound) { + return false + } + + var statusErr *httpStatusError + if errors.As(err, &statusErr) { + return isRetryableStatusCode(statusErr.statusCode) + } + + switch { + case errors.Is(err, errShortRead), + errors.Is(err, io.ErrUnexpectedEOF), + errors.Is(err, io.EOF), + errors.Is(err, syscall.ECONNRESET), + errors.Is(err, syscall.EPIPE), + errors.Is(err, syscall.ETIMEDOUT): + return true + } + + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return true + } + + msg := strings.ToLower(err.Error()) + for _, transient := range transientTransportMessages { + if strings.Contains(msg, transient) { + return true + } + } + return false +} + +// redactURLQuery strips the query string so that the signed download token never +// reaches stdout or a log aggregator. +func redactURLQuery(rawURL string) string { + parsed, err := url.Parse(rawURL) + if err != nil { + return rawURL + } + parsed.RawQuery = "" + parsed.Fragment = "" + return parsed.String() +} + +// IsTransientDownloadError reports whether err is a transient plugin download +// failure - a network or server-side problem rather than a bad plugin reference. +// Callers use it to keep advice about plugin resolution off errors that have +// nothing to do with it. +func IsTransientDownloadError(err error) bool { + return isRetryableDownloadError(err) +} diff --git a/managedplugin/download_retry_test.go b/managedplugin/download_retry_test.go new file mode 100644 index 0000000..2b9b7ed --- /dev/null +++ b/managedplugin/download_retry_test.go @@ -0,0 +1,184 @@ +package managedplugin + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "syscall" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestIsRetryableDownloadError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {name: "nil", err: nil, want: false}, + { + name: "http2 mid-body internal error stream reset", + err: fmt.Errorf("failed to copy body to file plugin.zip: %w", errors.New("stream error: stream ID 1; INTERNAL_ERROR; received from peer")), + want: true, + }, + { + name: "http2 refused stream", + err: errors.New("stream error: stream ID 3; REFUSED_STREAM"), + want: true, + }, + { + name: "http2 goaway", + err: errors.New("http2: server sent GOAWAY and closed the connection"), + want: true, + }, + {name: "connection reset", err: fmt.Errorf("read tcp: %w", syscall.ECONNRESET), want: true}, + {name: "broken pipe", err: fmt.Errorf("write tcp: %w", syscall.EPIPE), want: true}, + {name: "unexpected EOF", err: fmt.Errorf("failed to copy body to file: %w", io.ErrUnexpectedEOF), want: true}, + {name: "bare EOF", err: io.EOF, want: true}, + {name: "short read", err: fmt.Errorf("%w: got 10 bytes, want 20", errShortRead), want: true}, + {name: "net timeout", err: &net.OpError{Op: "read", Err: os.ErrDeadlineExceeded}, want: true}, + + {name: "expired signed URL 403", err: &httpStatusError{statusCode: http.StatusForbidden}, want: false}, + {name: "unauthorized 401", err: &httpStatusError{statusCode: http.StatusUnauthorized}, want: false}, + {name: "not found sentinel", err: errNotFound, want: false}, + {name: "not found 404 status", err: &httpStatusError{statusCode: http.StatusNotFound}, want: false}, + {name: "bad request 400", err: &httpStatusError{statusCode: http.StatusBadRequest}, want: false}, + + {name: "request timeout 408", err: &httpStatusError{statusCode: http.StatusRequestTimeout}, want: true}, + {name: "too many requests 429", err: &httpStatusError{statusCode: http.StatusTooManyRequests}, want: true}, + {name: "bad gateway 502", err: &httpStatusError{statusCode: http.StatusBadGateway}, want: true}, + {name: "service unavailable 503", err: &httpStatusError{statusCode: http.StatusServiceUnavailable}, want: true}, + + {name: "context canceled", err: fmt.Errorf("get url: %w", context.Canceled), want: false}, + {name: "context deadline exceeded", err: fmt.Errorf("get url: %w", context.DeadlineExceeded), want: false}, + {name: "checksum mismatch is permanent", err: errors.New("checksum mismatch: expected abc, got def"), want: false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isRetryableDownloadError(tc.err)) + }) + } +} + +func TestRedactURLQuery(t *testing.T) { + require.Equal(t, + "https://assets.cloudquery.io/cq-cloud-releases/cloudquery/source/envzero/v2.1.0/linux_amd64", + redactURLQuery("https://assets.cloudquery.io/cq-cloud-releases/cloudquery/source/envzero/v2.1.0/linux_amd64?verify=1787270563-PVHLM7Vfma5I0Mzx1YZXt4hPqXydx5WrqxPQf7B80I8%3D"), + ) +} + +// TestDownloadFileRetriesTruncatedBody reproduces the production failure: the first +// attempt writes part of the body and then the connection drops mid-copy. The retry +// must start the file from scratch rather than append to the partial bytes. +func TestDownloadFileRetriesTruncatedBody(t *testing.T) { + body := []byte("cloudquery-plugin-binary-payload") + + var attempts int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts++ + if attempts == 1 { + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body))) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body[:5]) + w.(http.Flusher).Flush() + // Close the connection mid-body so the client sees a truncated response. + conn, _, err := w.(http.Hijacker).Hijack() + require.NoError(t, err) + conn.Close() + return + } + _, _ = w.Write(body) + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.NoError(t, err) + require.Equal(t, 2, attempts) + + written, err := os.ReadFile(localPath) + require.NoError(t, err) + require.Equal(t, body, written, "retry must not append to the partial first attempt") + require.Equal(t, sha256Hex(body), checksum) +} + +func TestDownloadFileDoesNotRetryExpiredSignedURL(t *testing.T) { + var attempts int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts++ + w.WriteHeader(http.StatusForbidden) + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + _, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.Error(t, err) + require.Equal(t, 1, attempts, "an expired signed URL must fail fast") + require.Contains(t, err.Error(), "statusCode 403") +} + +func TestDownloadFileDoesNotRetryNotFound(t *testing.T) { + var attempts int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts++ + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + _, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.ErrorIs(t, err, errNotFound) + require.Equal(t, 1, attempts) +} + +func TestDownloadFileRetriesServerError(t *testing.T) { + body := []byte("payload") + + var attempts int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts++ + if attempts < 3 { + w.WriteHeader(http.StatusBadGateway) + return + } + _, _ = w.Write(body) + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + checksum, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.NoError(t, err) + require.Equal(t, 3, attempts) + require.Equal(t, sha256Hex(body), checksum) +} + +func TestDownloadFileGivesUpAfterRetryAttempts(t *testing.T) { + var attempts int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts++ + w.WriteHeader(http.StatusServiceUnavailable) + })) + t.Cleanup(server.Close) + + localPath := filepath.Join(t.TempDir(), "plugin.zip") + _, err := downloadFile(context.Background(), localPath, server.URL, DownloaderOptions{NoProgress: true}) + require.Error(t, err) + require.Equal(t, RetryAttempts, attempts) + require.Contains(t, err.Error(), "failed downloading URL") + require.NotContains(t, err.Error(), "verify=", "the signed token must not reach the error message") +} + +func sha256Hex(b []byte) string { + s := sha256.New() + s.Write(b) + return fmt.Sprintf("%x", s.Sum(nil)) +} From cded9c06a79ea46c800cc90e5e4a04bcdba30fab Mon Sep 17 00:00:00 2001 From: erezrokah Date: Fri, 21 Aug 2026 20:21:30 +0100 Subject: [PATCH 2/2] chore: Move the atomic binary extraction out to a follow-up PR --- managedplugin/download.go | 66 ++++++++++++++++++++------------------- 1 file changed, 34 insertions(+), 32 deletions(-) diff --git a/managedplugin/download.go b/managedplugin/download.go index e989285..86ad761 100644 --- a/managedplugin/download.go +++ b/managedplugin/download.go @@ -190,8 +190,6 @@ func doDownloadPluginFromHub(ctx context.Context, logger zerolog.Logger, c *clou return errors.New("failed to get plugin metadata from hub: empty location from response") } pluginZipPath := ops.LocalPath + ".zip" - defer os.Remove(pluginZipPath) - writtenChecksum, err := downloadFile(ctx, pluginZipPath, location, dops) if err != nil { return fmt.Errorf("failed to download plugin: %w", err) @@ -203,46 +201,29 @@ func doDownloadPluginFromHub(ctx context.Context, logger zerolog.Logger, c *clou return fmt.Errorf("checksum mismatch: expected %s, got %s", pluginAsset.Checksum, writtenChecksum) } - pathInArchive := fmt.Sprintf("plugin-%s-%s-%s-%s", ops.PluginName, ops.PluginVersion, runtime.GOOS, runtime.GOARCH) - return extractPluginBinary(pluginZipPath, pathInArchive, ops.LocalPath) -} - -// extractPluginBinary writes the binary to a temporary file and renames it into -// place, so a failure part way through never leaves a truncated binary that the -// next run treats as a cached plugin. -func extractPluginBinary(archivePath, pathInArchive, localPath string) error { - archive, err := zip.OpenReader(archivePath) + archive, err := zip.OpenReader(pluginZipPath) if err != nil { return fmt.Errorf("failed to open plugin archive: %w", err) } defer archive.Close() - fileInArchive, err := archive.Open(pathInArchive) + fileInArchive, err := archive.Open(fmt.Sprintf("plugin-%s-%s-%s-%s", ops.PluginName, ops.PluginVersion, runtime.GOOS, runtime.GOARCH)) if err != nil { - return fmt.Errorf("failed to open plugin archive %s: %w", pathInArchive, err) + return fmt.Errorf("failed to open plugin archive: %w", err) } - defer fileInArchive.Close() - out, err := os.CreateTemp(filepath.Dir(localPath), filepath.Base(localPath)+".tmp") + out, err := os.OpenFile(ops.LocalPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0744) if err != nil { - return fmt.Errorf("failed to create file %s: %w", localPath, err) + return fmt.Errorf("failed to create file %s: %w", ops.LocalPath, err) } - tmpPath := out.Name() - defer os.Remove(tmpPath) - - if _, err := io.Copy(out, fileInArchive); err != nil { - out.Close() + _, err = io.Copy(out, fileInArchive) + if err != nil { return fmt.Errorf("failed to copy body to file: %w", err) } - if err := out.Close(); err != nil { + err = out.Close() + if err != nil { return fmt.Errorf("failed to close file: %w", err) } - if err := os.Chmod(tmpPath, 0744); err != nil { - return fmt.Errorf("failed to set permissions on %s: %w", localPath, err) - } - if err := os.Rename(tmpPath, localPath); err != nil { - return fmt.Errorf("failed to move plugin binary to %s: %w", localPath, err) - } return nil } @@ -302,13 +283,17 @@ func doDownloadPluginFromGithub(ctx context.Context, logger zerolog.Logger, loca if err != nil { return fmt.Errorf("failed to get plugin url: %w", err) } - logger.Debug().Msg(fmt.Sprintf("Downloading %s", redactURLQuery(downloadURL))) - defer os.Remove(pluginZipPath) - + logger.Debug().Msg(fmt.Sprintf("Downloading %s", downloadURL)) if _, err := downloadFile(ctx, pluginZipPath, downloadURL, dops); err != nil { return fmt.Errorf("failed to download plugin: %w", err) } + archive, err := zip.OpenReader(pluginZipPath) + if err != nil { + return fmt.Errorf("failed to open plugin archive: %w", err) + } + defer archive.Close() + var pathInArchive string switch { case strings.HasPrefix(downloadURL, "https://github.com/cloudquery/cloudquery/releases/download/plugins-plugin"): @@ -327,7 +312,24 @@ func doDownloadPluginFromGithub(ctx context.Context, logger zerolog.Logger, loca return fmt.Errorf("unknown GitHub %s", downloadURL) } - return extractPluginBinary(pluginZipPath, WithBinarySuffix(pathInArchive), localPath) + pathInArchive = WithBinarySuffix(pathInArchive) + fileInArchive, err := archive.Open(pathInArchive) + if err != nil { + return fmt.Errorf("failed to open plugin archive plugins/source/%s: %w", name, err) + } + out, err := os.OpenFile(localPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0744) + if err != nil { + return fmt.Errorf("failed to create file %s: %w", localPath, err) + } + _, err = io.Copy(out, fileInArchive) + if err != nil { + return fmt.Errorf("failed to copy body to file: %w", err) + } + err = out.Close() + if err != nil { + return fmt.Errorf("failed to close file: %w", err) + } + return nil } func downloadFile(ctx context.Context, localPath string, downloadURL string, dops DownloaderOptions) (string, error) {