From 9f496c469fb0674eb11385afa307cceb09248993 Mon Sep 17 00:00:00 2001 From: actiontech-zihan Date: Wed, 12 Aug 2026 02:24:49 +0000 Subject: [PATCH] fix: use api enable_https for export download node proxy Cross-node data export downloads always built http:// targets from Request.Proto (HTTP version), so HTTPS clusters failed with 400. Derive scheme from dms.api.enable_https, skip TLS verify for internal proxy, and add download proxy logs plus unit tests. --- .../service/data_export_download_test.go | 157 ++++++++++++++++++ internal/apiserver/service/dms_controller.go | 40 ++++- 2 files changed, 193 insertions(+), 4 deletions(-) create mode 100644 internal/apiserver/service/data_export_download_test.go diff --git a/internal/apiserver/service/data_export_download_test.go b/internal/apiserver/service/data_export_download_test.go new file mode 100644 index 00000000..4a9cd06e --- /dev/null +++ b/internal/apiserver/service/data_export_download_test.go @@ -0,0 +1,157 @@ +package service + +import ( + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + utilLog "github.com/actiontech/dms/pkg/dms-common/pkg/log" + "github.com/labstack/echo/v4" +) + +type nopLogger struct{} + +func (nopLogger) Log(_ utilLog.Level, _ ...interface{}) error { return nil } + +func newTestDownloadController(enableHttps bool) *DMSController { + return &DMSController{ + log: utilLog.NewHelper(nopLogger{}, utilLog.WithMessageKey("test")), + enableHttps: enableHttps, + } +} + +func TestNodeProxyScheme(t *testing.T) { + tests := []struct { + enableHttps bool + want string + }{ + {enableHttps: true, want: "https"}, + {enableHttps: false, want: "http"}, + } + for _, tt := range tests { + t.Run(tt.want, func(t *testing.T) { + if got := nodeProxyScheme(tt.enableHttps); got != tt.want { + t.Fatalf("nodeProxyScheme(%v) = %q, want %q", tt.enableHttps, got, tt.want) + } + }) + } +} + +func serveExportFile(body string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set(echo.HeaderContentDisposition, `attachment; filename="export.zip"`) + w.Header().Set(echo.HeaderContentType, "application/zip") + _, _ = io.WriteString(w, body) + }) +} + +func invokeProxyDownload(t *testing.T, ctl *DMSController, reportHost string) (*httptest.ResponseRecorder, error) { + t.Helper() + e := echo.New() + req := httptest.NewRequest(http.MethodGet, "/v1/dms/projects/p1/data_export_tasks/t1/download", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + err := ctl.proxyDownloadDataExportTask(c, reportHost) + return rec, err +} + +func reportHostFromServerURL(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + panic(err) + } + return u.Host +} + +func TestProxyDownloadDataExportTaskHTTPSTarget(t *testing.T) { + const body = "export-zip-bytes" + server := httptest.NewTLSServer(serveExportFile(body)) + defer server.Close() + + ctl := newTestDownloadController(true) + rec, err := invokeProxyDownload(t, ctl, reportHostFromServerURL(server.URL)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d, body=%q", rec.Code, http.StatusOK, rec.Body.String()) + } + if got := rec.Body.String(); got != body { + t.Fatalf("body = %q, want %q", got, body) + } + if !strings.Contains(rec.Header().Get(echo.HeaderContentDisposition), "export.zip") { + t.Fatalf("missing content-disposition, got %q", rec.Header().Get(echo.HeaderContentDisposition)) + } +} + +func TestProxyDownloadDataExportTaskHTTPTarget(t *testing.T) { + const body = "export-zip-http" + server := httptest.NewServer(serveExportFile(body)) + defer server.Close() + + ctl := newTestDownloadController(false) + rec, err := invokeProxyDownload(t, ctl, reportHostFromServerURL(server.URL)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d, body=%q", rec.Code, http.StatusOK, rec.Body.String()) + } + if got := rec.Body.String(); got != body { + t.Fatalf("body = %q, want %q", got, body) + } +} + +func TestProxyDownloadDataExportTaskSchemeMismatch(t *testing.T) { + const body = "should-not-receive" + server := httptest.NewTLSServer(serveExportFile(body)) + defer server.Close() + + // Old bug: always used http against an HTTPS peer. + ctl := newTestDownloadController(false) + rec, err := invokeProxyDownload(t, ctl, reportHostFromServerURL(server.URL)) + if err == nil && rec.Code == http.StatusOK && rec.Body.String() == body { + t.Fatal("expected scheme mismatch to fail download, but got successful response") + } + if err != nil { + httpErr, ok := err.(*echo.HTTPError) + if !ok { + t.Fatalf("error type = %T, want *echo.HTTPError", err) + } + if httpErr.Code != http.StatusBadGateway { + t.Fatalf("error code = %d, want %d", httpErr.Code, http.StatusBadGateway) + } + return + } + // Plain HTTP against a Go TLS listener typically yields HTTP/1.0 400. + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d or transport error; body=%q", rec.Code, http.StatusBadRequest, rec.Body.String()) + } +} + +func TestProxyDownloadDataExportTaskUnreachable(t *testing.T) { + server := httptest.NewServer(serveExportFile("unused")) + reportHost := reportHostFromServerURL(server.URL) + server.Close() + + ctl := newTestDownloadController(false) + _, err := invokeProxyDownload(t, ctl, reportHost) + if err == nil { + t.Fatal("expected error for unreachable target") + } + httpErr, ok := err.(*echo.HTTPError) + if !ok { + t.Fatalf("error type = %T, want *echo.HTTPError", err) + } + if httpErr.Code != http.StatusBadGateway { + t.Fatalf("error code = %d, want %d", httpErr.Code, http.StatusBadGateway) + } + msg := fmt.Sprint(httpErr.Message) + if !strings.Contains(msg, "could not forward") { + t.Fatalf("error message = %q, want contain %q", msg, "could not forward") + } +} diff --git a/internal/apiserver/service/dms_controller.go b/internal/apiserver/service/dms_controller.go index fc5db984..baf13e75 100644 --- a/internal/apiserver/service/dms_controller.go +++ b/internal/apiserver/service/dms_controller.go @@ -3,6 +3,7 @@ package service import ( "bytes" "context" + "crypto/tls" "encoding/json" "errors" "fmt" @@ -40,7 +41,8 @@ type DMSController struct { DMS *service.DMSService CloudbeaverService *service.CloudbeaverService - log *utilLog.Helper + log *utilLog.Helper + enableHttps bool shutdownCallback func() error } @@ -49,11 +51,16 @@ func NewDMSController(logger utilLog.Logger, opts *conf.DMSOptions, cbService *s if nil != err { return nil, fmt.Errorf("failed to init dms service: %v", err) } + enableHttps := false + if opts != nil && opts.APIServiceOpts != nil { + enableHttps = opts.APIServiceOpts.EnableHttps + } return &DMSController{ // log: log.NewHelper(log.With(logger, "module", "controller/DMS")), DMS: dmsService, CloudbeaverService: cbService, - log: utilLog.NewHelper(logger, utilLog.WithMessageKey("controller")), + log: utilLog.NewHelper(logger, utilLog.WithMessageKey("controller")), + enableHttps: enableHttps, shutdownCallback: func() error { if err := dmsService.Shutdown(); err != nil { return err @@ -4203,13 +4210,16 @@ func (ctl *DMSController) DownloadDataExportTask(c echo.Context) error { isProxy, filePath, err := ctl.DMS.DownloadDataExportTask(c.Request().Context(), req, currentUserUid) if nil != err { + ctl.log.Errorf("DownloadDataExportTask failed: task_uid=%s user_uid=%s error=%v", req.DataExportTaskUid, currentUserUid, err) return NewErrResp(c, err, apiError.DMSServiceErr) } if isProxy { + ctl.log.Infof("DownloadDataExportTask proxy: task_uid=%s user_uid=%s report_host=%s", req.DataExportTaskUid, currentUserUid, filePath) return ctl.proxyDownloadDataExportTask(c, filePath) } + ctl.log.Infof("DownloadDataExportTask local: task_uid=%s user_uid=%s file=%s", req.DataExportTaskUid, currentUserUid, filePath) fileName := filepath.Base(filePath) c.Response().Header().Set(echo.HeaderContentDisposition, mime.FormatMediaType("attachment", map[string]string{"filename": fileName})) @@ -4217,12 +4227,32 @@ func (ctl *DMSController) DownloadDataExportTask(c echo.Context) error { return c.File(filePath) } +// nodeProxyScheme returns the scheme used for inter-node reverse proxy. +// Cluster nodes are assumed to share the same dms.api.enable_https setting. +func nodeProxyScheme(enableHttps bool) string { + if enableHttps { + return "https" + } + return "http" +} + func (ctl *DMSController) proxyDownloadDataExportTask(c echo.Context, reportHost string) (err error) { - protocol := strings.ToLower(strings.Split(c.Request().Proto, "/")[0]) + scheme := nodeProxyScheme(ctl.enableHttps) + target, err := url.Parse(fmt.Sprintf("%s://%s", scheme, reportHost)) + if err != nil { + ctl.log.Errorf("proxyDownloadDataExportTask invalid target: report_host=%s scheme=%s error=%v", reportHost, scheme, err) + return echo.NewHTTPError(http.StatusBadGateway, fmt.Sprintf("invalid report host %s: %v", reportHost, err)) + } + ctl.log.Infof("proxyDownloadDataExportTask target=%s", target.String()) // reference from echo framework proxy middleware - target, _ := url.Parse(fmt.Sprintf("%s://%s", protocol, reportHost)) reverseProxy := httputil.NewSingleHostReverseProxy(target) + if scheme == "https" { + // Inter-node traffic is internal; certificates are often self-signed without IP SANs. + reverseProxy.Transport = &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // cluster internal proxy + } + } reverseProxy.ErrorHandler = func(resp http.ResponseWriter, req *http.Request, err error) { // If the client canceled the request (usually by closing the connection), we can report a // client error (4xx) instead of a server error (5xx) to correctly identify the situation. @@ -4230,10 +4260,12 @@ func (ctl *DMSController) proxyDownloadDataExportTask(c echo.Context, reportHost // context.Canceled error with unexported garbage value requiring a substring check, see // https://github.com/golang/go/blob/6965b01ea248cabb70c3749fd218b36089a21efb/src/net/net.go#L416-L430 if err == context.Canceled || strings.Contains(err.Error(), "operation was canceled") { + ctl.log.Errorf("proxyDownloadDataExportTask client closed connection: target=%s error=%v", target.String(), err) httpError := echo.NewHTTPError(middleware.StatusCodeContextCanceled, fmt.Sprintf("client closed connection: %v", err)) httpError.Internal = err c.Set("_error", httpError) } else { + ctl.log.Errorf("proxyDownloadDataExportTask unreachable: target=%s error=%v", target.String(), err) httpError := echo.NewHTTPError(http.StatusBadGateway, fmt.Sprintf("remote %s unreachable, could not forward: %v", reportHost, err)) httpError.Internal = err c.Set("_error", httpError)