package handler import ( "context" "encoding/json" "io" "net/http" "strconv" "time" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/response" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/tokens" "git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer" "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" ) const ( jsonHeader = "application/json; charset=UTF-8" drainBufSize = 4096 ) type putResponse struct { ObjectID string `json:"object_id"` ContainerID string `json:"container_id"` } func newPutResponse(addr oid.Address) *putResponse { return &putResponse{ ObjectID: addr.Object().EncodeToString(), ContainerID: addr.Container().EncodeToString(), } } func (pr *putResponse) encode(w io.Writer) error { enc := json.NewEncoder(w) enc.SetIndent("", "\t") return enc.Encode(pr) } // Upload handles multipart upload request. func (h *Handler) Upload(req *fasthttp.RequestCtx) { var ( file MultipartFile idObj oid.ID addr oid.Address scid, _ = req.UserValue("cid").(string) log = h.log.With(zap.String("cid", scid)) bodyStream = req.RequestBodyStream() drainBuf = make([]byte, drainBufSize) ) ctx := utils.GetContextFromRequest(req) bktInfo, err := h.getBucketInfo(ctx, scid, log) if err != nil { logAndSendBucketError(req, log, err) return } defer func() { // If the temporary reader can be closed - let's close it. if file == nil { return } err := file.Close() log.Debug( logs.CloseTemporaryMultipartFormFile, zap.Stringer("address", addr), zap.String("filename", file.FileName()), zap.Error(err), ) }() boundary := string(req.Request.Header.MultipartFormBoundary()) if file, err = fetchMultipartFile(h.log, bodyStream, boundary); err != nil { log.Error(logs.CouldNotReceiveMultipartForm, zap.Error(err)) response.Error(req, "could not receive multipart/form: "+err.Error(), fasthttp.StatusBadRequest) return } filtered, err := filterHeaders(h.log, &req.Request.Header) if err != nil { log.Error(logs.CouldNotProcessHeaders, zap.Error(err)) response.Error(req, err.Error(), fasthttp.StatusBadRequest) return } now := time.Now() if rawHeader := req.Request.Header.Peek(fasthttp.HeaderDate); rawHeader != nil { if parsed, err := time.Parse(http.TimeFormat, string(rawHeader)); err != nil { log.Warn(logs.CouldNotParseClientTime, zap.String("Date header", string(rawHeader)), zap.Error(err)) } else { now = parsed } } if err = utils.PrepareExpirationHeader(req, h.frostfs, filtered, now); err != nil { log.Error(logs.CouldNotPrepareExpirationHeader, zap.Error(err)) response.Error(req, "could not prepare expiration header: "+err.Error(), fasthttp.StatusBadRequest) return } attributes := make([]object.Attribute, 0, len(filtered)) // prepares attributes from filtered headers for key, val := range filtered { attribute := object.NewAttribute() attribute.SetKey(key) attribute.SetValue(val) attributes = append(attributes, *attribute) } // sets FileName attribute if it wasn't set from header if _, ok := filtered[object.AttributeFileName]; !ok { filename := object.NewAttribute() filename.SetKey(object.AttributeFileName) filename.SetValue(file.FileName()) attributes = append(attributes, *filename) } // sets Timestamp attribute if it wasn't set from header and enabled by settings if _, ok := filtered[object.AttributeTimestamp]; !ok && h.config.DefaultTimestamp() { timestamp := object.NewAttribute() timestamp.SetKey(object.AttributeTimestamp) timestamp.SetValue(strconv.FormatInt(time.Now().Unix(), 10)) attributes = append(attributes, *timestamp) } obj := object.New() obj.SetContainerID(bktInfo.CID) obj.SetOwnerID(*h.ownerID) obj.SetAttributes(attributes...) prm := PrmObjectCreate{ PrmAuth: PrmAuth{ BearerToken: h.fetchBearerToken(ctx), }, Object: obj, Payload: file, ClientCut: h.config.ClientCut(), WithoutHomomorphicHash: bktInfo.HomomorphicHashDisabled, BufferMaxSize: h.config.BufferMaxSizeForPut(), } if idObj, err = h.frostfs.CreateObject(ctx, prm); err != nil { h.handlePutFrostFSErr(req, err) return } addr.SetObject(idObj) addr.SetContainer(bktInfo.CID) // Try to return the response, otherwise, if something went wrong, throw an error. if err = newPutResponse(addr).encode(req); err != nil { log.Error(logs.CouldNotEncodeResponse, zap.Error(err)) response.Error(req, "could not encode response", fasthttp.StatusBadRequest) return } // Multipart is multipart and thus can contain more than one part which // we ignore at the moment. Also, when dealing with chunked encoding // the last zero-length chunk might be left unread (because multipart // reader only cares about its boundary and doesn't look further) and // it will be (erroneously) interpreted as the start of the next // pipelined header. Thus we need to drain the body buffer. for { _, err = bodyStream.Read(drainBuf) if err == io.EOF || err == io.ErrUnexpectedEOF { break } } // Report status code and content type. req.Response.SetStatusCode(fasthttp.StatusOK) req.Response.Header.SetContentType(jsonHeader) } func (h *Handler) handlePutFrostFSErr(r *fasthttp.RequestCtx, err error) { statusCode, msg, additionalFields := response.FormErrorResponse("could not store file in frostfs", err) logFields := append([]zap.Field{zap.Error(err)}, additionalFields...) h.log.Error(logs.CouldNotStoreFileInFrostfs, logFields...) response.Error(r, msg, statusCode) } func (h *Handler) fetchBearerToken(ctx context.Context) *bearer.Token { if tkn, err := tokens.LoadBearerToken(ctx); err == nil && tkn != nil { return tkn } return nil }