package handler import ( "archive/tar" "archive/zip" "bufio" "compress/gzip" "context" "errors" "fmt" "io" "net/url" "strings" "time" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/tree" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils" "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "github.com/valyala/fasthttp" "go.uber.org/zap" ) // DownloadByAddressOrBucketName handles download requests using simple cid/oid or bucketname/key format. func (h *Handler) DownloadByAddressOrBucketName(req *fasthttp.RequestCtx) { ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadByAddressOrBucketName") defer span.End() cidParam := req.UserValue("cid").(string) oidParam := req.UserValue("oid").(string) ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With( zap.String("cid", cidParam), zap.String("oid", oidParam), )) path, err := url.QueryUnescape(oidParam) if err != nil { h.logAndSendError(ctx, req, logs.FailedToUnescapePath, err) return } bktInfo, err := h.getBucketInfo(ctx, cidParam) if err != nil { h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err) return } checkS3Err := h.tree.CheckSettingsNodeExists(ctx, bktInfo) if checkS3Err != nil && !errors.Is(checkS3Err, tree.ErrNodeNotFound) { h.logAndSendError(ctx, req, logs.FailedToCheckIfSettingsNodeExist, checkS3Err) return } prm := MiddlewareParam{ Context: ctx, Request: req, BktInfo: bktInfo, Path: path, } indexPageEnabled := h.config.IndexPageEnabled() if checkS3Err == nil { run(prm, h.errorMiddleware(logs.ObjectNotFound, ErrObjectNotFound), Middleware{Func: h.byS3PathMiddleware(h.receiveFile, noopFormer), Enabled: true}, Middleware{Func: h.byS3PathMiddleware(h.receiveFile, indexFormer), Enabled: indexPageEnabled}, Middleware{Func: h.browseIndexMiddleware(h.getDirObjectsS3), Enabled: indexPageEnabled}, ) } else { slashFallbackEnabled := h.config.EnableFilepathSlashFallback() fileNameFallbackEnabled := h.config.EnableFilepathFallback() run(prm, h.errorMiddleware(logs.ObjectNotFound, ErrObjectNotFound), Middleware{Func: h.byAddressMiddleware(h.receiveFile), Enabled: true}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, noopFormer), Enabled: true}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, reverseLeadingSlash), Enabled: slashFallbackEnabled}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, noopFormer), Enabled: fileNameFallbackEnabled}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, reverseLeadingSlash), Enabled: fileNameFallbackEnabled && slashFallbackEnabled}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, indexFormer), Enabled: indexPageEnabled}, Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, indexFormer), Enabled: fileNameFallbackEnabled && indexPageEnabled}, Middleware{Func: h.browseIndexMiddleware(h.getDirObjectsNative), Enabled: indexPageEnabled}, ) } } type MiddlewareFunc func(param MiddlewareParam) bool type MiddlewareParam struct { Context context.Context Request *fasthttp.RequestCtx BktInfo *data.BucketInfo Path string } type Middleware struct { Func MiddlewareFunc Enabled bool } func run(prm MiddlewareParam, defaultMiddleware MiddlewareFunc, middlewares ...Middleware) { for _, m := range middlewares { if m.Enabled && !m.Func(prm) { return } } defaultMiddleware(prm) } func indexFormer(path string) string { indexPath := path if indexPath != "" && !strings.HasSuffix(indexPath, "/") { indexPath += "/" } return indexPath + "index.html" } func reverseLeadingSlash(path string) string { if path == "" || path == "/" { return path } if path[0] == '/' { return path[1:] } return "/" + path } func noopFormer(path string) string { return path } func (h *Handler) byS3PathMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address), pathFormer func(string) string) MiddlewareFunc { return func(prm MiddlewareParam) bool { ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byS3Path") defer span.End() path := pathFormer(prm.Path) foundOID, err := h.tree.GetLatestVersion(ctx, &prm.BktInfo.CID, path) if err == nil { if foundOID.IsDeleteMarker { h.logAndSendError(ctx, prm.Request, logs.IndexWasDeleted, ErrObjectNotFound) return false } addr := newAddress(prm.BktInfo.CID, foundOID.OID) handler(ctx, prm.Request, addr) return false } if !errors.Is(err, tree.ErrNodeNotFound) { h.logAndSendError(ctx, prm.Request, logs.FailedToGetLatestVersionOfIndexObject, err, zap.String("path", path)) return false } return true } } func (h *Handler) byAttributeSearchMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address), attr string, pathFormer func(string) string) MiddlewareFunc { return func(prm MiddlewareParam) bool { ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byAttributeSearch") defer span.End() path := pathFormer(prm.Path) res, err := h.search(ctx, prm.BktInfo.CID, attr, path, object.MatchStringEqual) if err != nil { h.logAndSendError(ctx, prm.Request, logs.FailedToFindObjectByAttribute, err) return false } defer res.Close() buf := make([]oid.ID, 1) n, err := res.Read(buf) if err == nil && n > 0 { addr := newAddress(prm.BktInfo.CID, buf[0]) handler(ctx, prm.Request, addr) return false } if !errors.Is(err, io.EOF) { h.logAndSendError(ctx, prm.Request, logs.FailedToFindObjectByAttribute, err) return false } return true } } func (h *Handler) byAddressMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address)) MiddlewareFunc { return func(prm MiddlewareParam) bool { ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byAddress") defer span.End() var objID oid.ID if objID.DecodeString(prm.Path) == nil { handler(ctx, prm.Request, newAddress(prm.BktInfo.CID, objID)) return false } return true } } // DownloadByAttribute handles attribute-based download requests. func (h *Handler) DownloadByAttribute(req *fasthttp.RequestCtx) { ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadByAttribute") defer span.End() h.byAttribute(ctx, req, h.receiveFile) } func (h *Handler) search(ctx context.Context, cnrID cid.ID, key, val string, op object.SearchMatchType) (ResObjectSearch, error) { filters := object.NewSearchFilters() filters.AddRootFilter() filters.AddFilter(key, val, op) prm := PrmObjectSearch{ PrmAuth: PrmAuth{ BearerToken: bearerToken(ctx), }, Container: cnrID, Filters: filters, } return h.frostfs.SearchObjects(ctx, prm) } // DownloadZip handles zip by prefix requests. func (h *Handler) DownloadZip(req *fasthttp.RequestCtx) { ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadZip") defer span.End() scid, _ := req.UserValue("cid").(string) prefix, _ := req.UserValue("prefix").(string) ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With(zap.String("cid", scid), zap.String("prefix", prefix))) bktInfo, err := h.getBucketInfo(ctx, scid) if err != nil { h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err) return } resSearch, err := h.searchObjectsByPrefix(ctx, bktInfo.CID, prefix) if err != nil { return } req.Response.Header.Set(fasthttp.HeaderContentType, "application/zip") req.Response.Header.Set(fasthttp.HeaderContentDisposition, "attachment; filename=\"archive.zip\"") req.SetBodyStreamWriter(h.getZipResponseWriter(ctx, resSearch, bktInfo)) } func (h *Handler) getZipResponseWriter(ctx context.Context, resSearch ResObjectSearch, bktInfo *data.BucketInfo) func(w *bufio.Writer) { return func(w *bufio.Writer) { defer resSearch.Close() buf := make([]byte, 3<<20) zipWriter := zip.NewWriter(w) var objectsWritten int errIter := resSearch.Iterate(h.putObjectToArchive(ctx, bktInfo.CID, buf, func(obj *object.Object) (io.Writer, error) { objectsWritten++ return h.createZipFile(zipWriter, obj) }), ) if errIter != nil { h.reqLogger(ctx).Error(logs.IteratingOverSelectedObjectsFailed, zap.Error(errIter), logs.TagField(logs.TagDatapath)) return } else if objectsWritten == 0 { h.reqLogger(ctx).Warn(logs.ObjectsNotFound, logs.TagField(logs.TagDatapath)) } if err := zipWriter.Close(); err != nil { h.reqLogger(ctx).Error(logs.CloseZipWriter, zap.Error(err), logs.TagField(logs.TagDatapath)) } } } func (h *Handler) createZipFile(zw *zip.Writer, obj *object.Object) (io.Writer, error) { method := zip.Store if h.config.ArchiveCompression() { method = zip.Deflate } filePath := getFilePath(obj) if len(filePath) == 0 || filePath[len(filePath)-1] == '/' { return nil, fmt.Errorf("invalid filepath '%s'", filePath) } return zw.CreateHeader(&zip.FileHeader{ Name: filePath, Method: method, Modified: time.Now(), }) } // DownloadTar forms tar.gz from objects by prefix. func (h *Handler) DownloadTar(req *fasthttp.RequestCtx) { ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadTar") defer span.End() scid, _ := req.UserValue("cid").(string) prefix, _ := req.UserValue("prefix").(string) ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With(zap.String("cid", scid), zap.String("prefix", prefix))) bktInfo, err := h.getBucketInfo(ctx, scid) if err != nil { h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err) return } resSearch, err := h.searchObjectsByPrefix(ctx, bktInfo.CID, prefix) if err != nil { return } req.Response.Header.Set(fasthttp.HeaderContentType, "application/gzip") req.Response.Header.Set(fasthttp.HeaderContentDisposition, "attachment; filename=\"archive.tar.gz\"") req.SetBodyStreamWriter(h.getTarResponseWriter(ctx, resSearch, bktInfo)) } func (h *Handler) getTarResponseWriter(ctx context.Context, resSearch ResObjectSearch, bktInfo *data.BucketInfo) func(w *bufio.Writer) { return func(w *bufio.Writer) { defer resSearch.Close() compressionLevel := gzip.NoCompression if h.config.ArchiveCompression() { compressionLevel = gzip.DefaultCompression } // ignore error because it's not nil only if compressionLevel argument is invalid gzipWriter, _ := gzip.NewWriterLevel(w, compressionLevel) tarWriter := tar.NewWriter(gzipWriter) defer func() { if err := tarWriter.Close(); err != nil { h.reqLogger(ctx).Error(logs.CloseTarWriter, zap.Error(err), logs.TagField(logs.TagDatapath)) } if err := gzipWriter.Close(); err != nil { h.reqLogger(ctx).Error(logs.CloseGzipWriter, zap.Error(err), logs.TagField(logs.TagDatapath)) } }() var objectsWritten int buf := make([]byte, 3<<20) // the same as for upload errIter := resSearch.Iterate(h.putObjectToArchive(ctx, bktInfo.CID, buf, func(obj *object.Object) (io.Writer, error) { objectsWritten++ return h.createTarFile(tarWriter, obj) }), ) if errIter != nil { h.reqLogger(ctx).Error(logs.IteratingOverSelectedObjectsFailed, zap.Error(errIter), logs.TagField(logs.TagDatapath)) } else if objectsWritten == 0 { h.reqLogger(ctx).Warn(logs.ObjectsNotFound, logs.TagField(logs.TagDatapath)) } } } func (h *Handler) createTarFile(tw *tar.Writer, obj *object.Object) (io.Writer, error) { filePath := getFilePath(obj) if len(filePath) == 0 || filePath[len(filePath)-1] == '/' { return nil, fmt.Errorf("invalid filepath '%s'", filePath) } return tw, tw.WriteHeader(&tar.Header{ Name: filePath, Mode: 0655, Size: int64(obj.PayloadSize()), }) } func (h *Handler) putObjectToArchive(ctx context.Context, cnrID cid.ID, buf []byte, createArchiveHeader func(obj *object.Object) (io.Writer, error)) func(id oid.ID) bool { return func(id oid.ID) bool { logger := h.reqLogger(ctx).With(zap.String("oid", id.EncodeToString())) prm := PrmObjectGet{ PrmAuth: PrmAuth{ BearerToken: bearerToken(ctx), }, Address: newAddress(cnrID, id), } resGet, err := h.frostfs.GetObject(ctx, prm) if err != nil { logger.Error(logs.FailedToGetObject, zap.Error(err), logs.TagField(logs.TagExternalStorage)) return false } fileWriter, err := createArchiveHeader(&resGet.Header) if err != nil { logger.Error(logs.FailedToAddObjectToArchive, zap.Error(err), logs.TagField(logs.TagDatapath)) return false } if err = writeToArchive(resGet, fileWriter, buf); err != nil { logger.Error(logs.FailedToAddObjectToArchive, zap.Error(err), logs.TagField(logs.TagDatapath)) return false } return false } } func (h *Handler) searchObjectsByPrefix(ctx context.Context, cnrID cid.ID, prefix string) (ResObjectSearch, error) { prefix, err := url.QueryUnescape(prefix) if err != nil { return nil, fmt.Errorf("unescape prefix: %w", err) } resSearch, err := h.search(ctx, cnrID, object.AttributeFilePath, prefix, object.MatchCommonPrefix) if err != nil { return nil, fmt.Errorf("search objects by prefix: %w", err) } return resSearch, nil } func writeToArchive(resGet *Object, objWriter io.Writer, buf []byte) error { var err error if _, err = io.CopyBuffer(objWriter, resGet.Payload, buf); err != nil { return fmt.Errorf("copy object payload to zip file: %v", err) } if err = resGet.Payload.Close(); err != nil { return fmt.Errorf("object body close error: %w", err) } return nil } func getFilePath(obj *object.Object) string { for _, attr := range obj.Attributes() { if attr.Key() == object.AttributeFilePath { return attr.Value() } } return "" }