package handler import ( "archive/tar" "archive/zip" "bufio" "compress/gzip" "context" "errors" "fmt" "io" "net/url" "time" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/layer" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs" "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) downloadParam := req.QueryArgs().GetBool("download") ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With( zap.String("cid", cidParam), zap.String("oid", oidParam), )) 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, layer.ErrNodeNotFound) { h.logAndSendError(ctx, req, logs.FailedToCheckIfSettingsNodeExist, checkS3Err) return } var objID oid.ID if checkS3Err == nil && shouldDownload(oidParam, downloadParam) { h.byS3Path(ctx, req, bktInfo.CID, oidParam, h.receiveFile) } else if err = objID.DecodeString(oidParam); err == nil { h.byNativeAddress(ctx, req, bktInfo.CID, objID, h.receiveFile) } else { h.browseIndex(ctx, req, cidParam, oidParam, checkS3Err != nil) } } func shouldDownload(oidParam string, downloadParam bool) bool { return !isDir(oidParam) || downloadParam } // 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 "" }