File upload with metadata (#282)

* Fixed merge conflicts

* merge conflict errors

* left a space

* fix review comments by Janos

* few more comments fix

* handle no content-length in multipart part bzzFileUploadHandler (#296)

* remove some redundant checks
Co-authored-by: Janoš Guljaš <janos@users.noreply.github.com>

pkg/api/bytes.go

@@ -11,14 +11,13 @@ import (
 	"io"
 	"net/http"
 
-	"github.com/gorilla/mux"
-
 	"github.com/ethersphere/bee/pkg/file"
 	"github.com/ethersphere/bee/pkg/file/joiner"
 	"github.com/ethersphere/bee/pkg/file/splitter"
 	"github.com/ethersphere/bee/pkg/jsonhttp"
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/swarm"
+	"github.com/gorilla/mux"
 )
 
 type bytesPostResponse struct {
@@ -31,8 +30,10 @@ func (s *server) bytesUploadHandler(w http.ResponseWriter, r *http.Request) {
 	responseObject, err := s.splitUpload(ctx, r.Body, r.ContentLength)
 	if err != nil {
 		s.Logger.Debugf("bytes upload: %v", err)
-		o := responseObject.(jsonhttp.StatusResponse)
-		jsonhttp.Respond(w, o.Code, o)
+		var response jsonhttp.StatusResponse
+		response.Message = "upload error"
+		response.Code = http.StatusInternalServerError
+		jsonhttp.Respond(w, response.Code, response)
 	} else {
 		jsonhttp.OK(w, responseObject)
 	}
@@ -61,12 +62,8 @@ func (s *server) splitUpload(ctx context.Context, r io.ReadCloser, l int64) (int
 	}()
 	sp := splitter.NewSimpleSplitter(s.Storer)
 	address, err := sp.Split(ctx, chunkPipe, l)
-	var response jsonhttp.StatusResponse
 	if err != nil {
-		response.Message = "upload error"
-		response.Code = http.StatusInternalServerError
-		err = fmt.Errorf("%s: %v", response.Message, err)
-		return response, err
+		return swarm.ZeroAddress, err
 	}
 	return bytesPostResponse{Reference: address}, nil
 }

pkg/api/file.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package api
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"mime"
+	"mime/multipart"
+	"net/http"
+	"os"
+	"strconv"
+
+	"github.com/ethersphere/bee/pkg/collection/entry"
+	"github.com/ethersphere/bee/pkg/file"
+	"github.com/ethersphere/bee/pkg/file/joiner"
+	"github.com/ethersphere/bee/pkg/jsonhttp"
+	"github.com/ethersphere/bee/pkg/storage"
+	"github.com/ethersphere/bee/pkg/swarm"
+	"github.com/gorilla/mux"
+)
+
+const (
+	MultiPartFormData = "multipart/form-data"
+)
+
+type FileUploadResponse struct {
+	Reference swarm.Address `json:"reference"`
+}
+
+// bzzFileUploadHandler uploads the file and its metadata supplied as a multipart http message.
+func (s *server) bzzFileUploadHandler(w http.ResponseWriter, r *http.Request) {
+	contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
+	if contentType != MultiPartFormData {
+		s.Logger.Debugf("file: no mutlipart: %v", err)
+		s.Logger.Error("file: no mutlipart")
+		jsonhttp.BadRequest(w, "not a mutlipart/form-data")
+		return
+	}
+
+	mr := multipart.NewReader(r.Body, params["boundary"])
+	for {
+		part, err := mr.NextPart()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			s.Logger.Debugf("file: read mutlipart: %v", err)
+			s.Logger.Error("file: read mutlipart")
+			jsonhttp.BadRequest(w, "error reading a mutlipart/form-data")
+			return
+		}
+
+		ctx := r.Context()
+
+		// try to find filename
+		// 1) in part header params
+		// 2) as formname
+		// 3) file reference hash (after uploading the file)
+		fileName := part.FileName()
+		if fileName == "" {
+			fileName = part.FormName()
+		}
+
+		var reader io.ReadCloser
+
+		// then find out content type
+		contentType := part.Header.Get("Content-Type")
+		if contentType == "" {
+			br := bufio.NewReader(part)
+			buf, err := br.Peek(512)
+			if err != nil && err != io.EOF {
+				s.Logger.Debugf("file: read content type: %v, file name %s", err, fileName)
+				s.Logger.Error("file: read content type")
+				jsonhttp.BadRequest(w, "error reading content type")
+				return
+			}
+			contentType = http.DetectContentType(buf)
+			reader = ioutil.NopCloser(br)
+		} else {
+			reader = part
+		}
+
+		var fileSize uint64
+		if contentLength := part.Header.Get("Content-Length"); contentLength != "" {
+			fileSize, err = strconv.ParseUint(contentLength, 10, 64)
+			if err != nil {
+				s.Logger.Debugf("file: content length: %v", err)
+				s.Logger.Error("file: content length")
+				jsonhttp.BadRequest(w, "invalid content length header")
+				return
+			}
+		} else {
+			// copy the part to a tmp file to get its size
+			tmp, err := ioutil.TempFile("", "bee-multipart")
+			if err != nil {
+				s.Logger.Debugf("file: create temporary file: %v", err)
+				s.Logger.Error("file: create temporary file")
+				jsonhttp.InternalServerError(w, nil)
+				return
+			}
+			defer os.Remove(tmp.Name())
+			defer tmp.Close()
+			n, err := io.Copy(tmp, part)
+			if err != nil {
+				s.Logger.Debugf("file: write temporary file: %v", err)
+				s.Logger.Error("file: write temporary file")
+				jsonhttp.InternalServerError(w, nil)
+				return
+			}
+			if _, err := tmp.Seek(0, io.SeekStart); err != nil {
+				s.Logger.Debugf("file: seek to beginning of temporary file: %v", err)
+				s.Logger.Error("file: seek to beginning of temporary file")
+				jsonhttp.InternalServerError(w, nil)
+				return
+			}
+			fileSize = uint64(n)
+			reader = tmp
+		}
+
+		// first store the file and get its reference
+		fr, err := s.storePartData(ctx, reader, fileSize)
+		if err != nil {
+			s.Logger.Debugf("file: file store: %v,", err)
+			s.Logger.Error("file: file store")
+			jsonhttp.InternalServerError(w, "could not store file data")
+			return
+		}
+
+		// If filename is still empty, use the file hash the filename
+		if fileName == "" {
+			fileName = fr.String()
+		}
+
+		// then store the metadata and get its reference
+		m := entry.NewMetadata(fileName)
+		m.MimeType = contentType
+		metadataBytes, err := json.Marshal(m)
+		if err != nil {
+			s.Logger.Debugf("file: metadata marshall: %v, file name %s", err, fileName)
+			s.Logger.Error("file: metadata marshall")
+			jsonhttp.InternalServerError(w, "metadata marshall error")
+			return
+		}
+		mr, err := s.storeMeta(ctx, metadataBytes)
+		if err != nil {
+			s.Logger.Debugf("file: metadata store: %v, file name %s", err, fileName)
+			s.Logger.Error("file: metadata store")
+			jsonhttp.InternalServerError(w, "could not store metadata")
+			return
+		}
+
+		// now join both references (mr,fr) to create an entry and store it.
+		entrie := entry.New(fr, mr)
+		fileEntryBytes, err := entrie.MarshalBinary()
+		if err != nil {
+			s.Logger.Debugf("file: entry marshall: %v, file name %s", err, fileName)
+			s.Logger.Error("file: entry marshall")
+			jsonhttp.InternalServerError(w, "entry marshall error")
+			return
+		}
+		er, err := s.storeMeta(ctx, fileEntryBytes)
+		if err != nil {
+			s.Logger.Debugf("file: entry store: %v, file name %s", err, fileName)
+			s.Logger.Error("file: entry store")
+			jsonhttp.InternalServerError(w, "could not store entry")
+			return
+		}
+		w.Header().Set("ETag", fmt.Sprintf("%q", er.String()))
+		jsonhttp.OK(w, &FileUploadResponse{Reference: er})
+	}
+}
+
+// bzzFileDownloadHandler downloads the file given the entry's reference.
+func (s *server) bzzFileDownloadHandler(w http.ResponseWriter, r *http.Request) {
+	addr := mux.Vars(r)["addr"]
+	address, err := swarm.ParseHexAddress(addr)
+	if err != nil {
+		s.Logger.Debugf("file: parse file address %s: %v", addr, err)
+		s.Logger.Error("file: parse file address")
+		jsonhttp.BadRequest(w, "invalid file address")
+		return
+	}
+
+	// read entry.
+	j := joiner.NewSimpleJoiner(s.Storer)
+	buf := bytes.NewBuffer(nil)
+	_, err = file.JoinReadAll(j, address, buf)
+	if err != nil {
+		s.Logger.Debugf("file: read entry %s: %v", addr, err)
+		s.Logger.Error("file: read entry")
+		jsonhttp.InternalServerError(w, "error reading entry")
+		return
+	}
+	e := &entry.Entry{}
+	err = e.UnmarshalBinary(buf.Bytes())
+	if err != nil {
+		s.Logger.Debugf("file: unmarshall entry %s: %v", addr, err)
+		s.Logger.Error("file: unmarshall entry")
+		jsonhttp.InternalServerError(w, "error unmarshalling entry")
+		return
+	}
+
+	// If none match header is set always send the reply as not modified
+	// TODO: when SOC comes, we need to revisit this concept
+	noneMatchEtag := r.Header.Get("If-None-Match")
+	if noneMatchEtag != "" {
+		if e.Reference().Equal(address) {
+			w.WriteHeader(http.StatusNotModified)
+			return
+		}
+	}
+
+	// Read metadata.
+	buf = bytes.NewBuffer(nil)
+	_, err = file.JoinReadAll(j, e.Metadata(), buf)
+	if err != nil {
+		s.Logger.Debugf("file: read metadata %s: %v", addr, err)
+		s.Logger.Error("file: read netadata")
+		jsonhttp.InternalServerError(w, "error reading metadata")
+		return
+	}
+	metaData := &entry.Metadata{}
+	err = json.Unmarshal(buf.Bytes(), metaData)
+	if err != nil {
+		s.Logger.Debugf("file: unmarshall metadata %s: %v", addr, err)
+		s.Logger.Error("file: unmarshall metadata")
+		jsonhttp.InternalServerError(w, "error unmarshalling metadata")
+		return
+	}
+
+	// send the file data back in the response
+	dataSize, err := j.Size(r.Context(), address)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			s.Logger.Debugf("file: not found %s: %v", address, err)
+			s.Logger.Error("file: not found")
+			jsonhttp.NotFound(w, "not found")
+			return
+		}
+		s.Logger.Debugf("file: invalid root chunk %s: %v", address, err)
+		s.Logger.Error("file: invalid root chunk")
+		jsonhttp.BadRequest(w, "invalid root chunk")
+		return
+	}
+
+	outBuffer := bytes.NewBuffer(nil)
+	c, err := file.JoinReadAll(j, e.Reference(), outBuffer)
+	if err != nil && c == 0 {
+		s.Logger.Debugf("file: data read %s: %v", addr, err)
+		s.Logger.Error("file: data read")
+		jsonhttp.InternalServerError(w, "error reading data")
+		return
+	}
+	w.Header().Set("ETag", fmt.Sprintf("%q", e.Reference()))
+	w.Header().Set("Content-Disposition", fmt.Sprintf("inline; filename=\"%s\"", metaData.Filename))
+	w.Header().Set("Content-Type", metaData.MimeType)
+	w.Header().Set("Content-Length", fmt.Sprintf("%d", dataSize))
+	_, _ = io.Copy(w, outBuffer)
+}
+
+// storeMeta is used to store metadata information as a whole.
+func (s *server) storeMeta(ctx context.Context, dataBytes []byte) (swarm.Address, error) {
+	dataBuf := bytes.NewBuffer(dataBytes)
+	dataReadCloser := ioutil.NopCloser(dataBuf)
+	o, err := s.splitUpload(ctx, dataReadCloser, int64(len(dataBytes)))
+	if err != nil {
+		return swarm.ZeroAddress, err
+	}
+	bytesResp := o.(bytesPostResponse)
+	return bytesResp.Reference, nil
+}
+
+// storePartData stores file data belonging to one of the part of multipart.
+func (s *server) storePartData(ctx context.Context, r io.ReadCloser, l uint64) (swarm.Address, error) {
+	o, err := s.splitUpload(ctx, r, int64(l))
+	if err != nil {
+		return swarm.ZeroAddress, err
+	}
+	bytesResp := o.(bytesPostResponse)
+	return bytesResp.Reference, nil
+}

pkg/api/file_test.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package api_test
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"mime"
+	"net/http"
+	"testing"
+
+	"github.com/ethersphere/bee/pkg/api"
+	"github.com/ethersphere/bee/pkg/jsonhttp"
+	"github.com/ethersphere/bee/pkg/jsonhttp/jsonhttptest"
+	"github.com/ethersphere/bee/pkg/logging"
+	"github.com/ethersphere/bee/pkg/storage/mock"
+	"github.com/ethersphere/bee/pkg/swarm"
+	"github.com/ethersphere/bee/pkg/tags"
+)
+
+func TestBzz(t *testing.T) {
+	var (
+		simpleResource  = func() string { return "/files" }
+		addressResource = func(addr string) string { return "/files/" + addr }
+		simpleData      = []byte("this is a simple text")
+		mockStorer      = mock.NewStorer()
+		client          = newTestServer(t, testServerOptions{
+			Storer: mockStorer,
+			Tags:   tags.NewTags(),
+			Logger: logging.New(ioutil.Discard, 5),
+		})
+	)
+	t.Run("simple-upload", func(t *testing.T) {
+		jsonhttptest.ResponseDirectSendHeadersAndReceiveHeaders(t, client, http.MethodPost, simpleResource(), bytes.NewReader(simpleData), http.StatusBadRequest, jsonhttp.StatusResponse{
+			Message: "not a mutlipart/form-data",
+			Code:    http.StatusBadRequest,
+		}, nil)
+	})
+
+	t.Run("simple-upload", func(t *testing.T) {
+		fileName := "simple_file.txt"
+		rootHash := "295673cf7aa55d119dd6f82528c91d45b53dd63dc2e4ca4abf4ed8b3a0788085"
+		_ = jsonhttptest.ResponseDirectWithMultiPart(t, client, http.MethodPost, simpleResource(), fileName, simpleData, http.StatusOK, "", api.FileUploadResponse{
+			Reference: swarm.MustParseHexAddress(rootHash),
+		})
+	})
+
+	t.Run("check-content-type-detection", func(t *testing.T) {
+		fileName := "my-pictures.jpeg"
+		rootHash := "f2e761160deda91c1fbfab065a5abf530b0766b3e102b51fbd626ba37c3bc581"
+		_ = jsonhttptest.ResponseDirectWithMultiPart(t, client, http.MethodPost, simpleResource(), fileName, simpleData, http.StatusOK, "image/jpeg; charset=utf-8", api.FileUploadResponse{
+			Reference: swarm.MustParseHexAddress(rootHash),
+		})
+
+		rcvdHeader := jsonhttptest.ResponseDirectCheckBinaryResponse(t, client, http.MethodGet, addressResource(rootHash), nil, http.StatusOK, simpleData, nil)
+		cd := rcvdHeader.Get("Content-Disposition")
+		_, params, err := mime.ParseMediaType(cd)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if params["filename"] != fileName {
+			t.Fatal("Invalid file name detected")
+		}
+		if rcvdHeader.Get("Content-Type") != "image/jpeg; charset=utf-8" {
+			t.Fatal("Invalid content type detected")
+		}
+	})
+
+	t.Run("upload-then-download-and-check-data", func(t *testing.T) {
+		fileName := "sample.html"
+		rootHash := "9f8ba407ff4809e877c75506247e0f1faf206262d1ddd7b3c8f9775d3501be50"
+		sampleHtml := `<!DOCTYPE html>
+		<html>
+		<body>
+
+		<h1>My First Heading</h1>
+
+		<p>My first paragraph.</p>
+
+		</body>
+		</html>`
+
+		rcvdHeader := jsonhttptest.ResponseDirectWithMultiPart(t, client, http.MethodPost, simpleResource(), fileName, []byte(sampleHtml), http.StatusOK, "", api.FileUploadResponse{
+			Reference: swarm.MustParseHexAddress(rootHash),
+		})
+
+		if rcvdHeader.Get("ETag") != fmt.Sprintf("%q", rootHash) {
+			t.Fatal("Invalid ETags header received")
+		}
+
+		// try to fetch the same file and check the data
+		rcvdHeader = jsonhttptest.ResponseDirectCheckBinaryResponse(t, client, http.MethodGet, addressResource(rootHash), nil, http.StatusOK, []byte(sampleHtml), nil)
+
+		// check the headers
+		cd := rcvdHeader.Get("Content-Disposition")
+		_, params, err := mime.ParseMediaType(cd)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if params["filename"] != fileName {
+			t.Fatal("Invalid filename detected")
+		}
+		if rcvdHeader.Get("Content-Type") != "text/html; charset=utf-8" {
+			t.Fatal("Invalid content type detected")
+		}
+	})
+}

pkg/api/router.go

@@ -38,6 +38,15 @@ func (s *server) setupRouting() {
 	handle(router, "/bytes", jsonhttp.MethodHandler{
 		"POST": http.HandlerFunc(s.bytesUploadHandler),
 	})
+
+	handle(router, "/files", jsonhttp.MethodHandler{
+		"POST": http.HandlerFunc(s.bzzFileUploadHandler),
+	})
+
+	handle(router, "/files/{addr}", jsonhttp.MethodHandler{
+		"GET": http.HandlerFunc(s.bzzFileDownloadHandler),
+	})
+
 	handle(router, "/bytes/{address}", jsonhttp.MethodHandler{
 		"GET": http.HandlerFunc(s.bytesGetHandler),
 	})

pkg/jsonhttp/jsonhttptest/jsonhttptest.go

@@ -7,9 +7,13 @@ package jsonhttptest
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"io"
 	"io/ioutil"
+	"mime/multipart"
 	"net/http"
+	"net/textproto"
+	"strconv"
 	"testing"
 )
 
@@ -35,6 +39,79 @@ func ResponseDirect(t *testing.T, client *http.Client, method, url string, body
 	}
 }
 
+func ResponseDirectWithMultiPart(t *testing.T, client *http.Client, method, url, fileName string, data []byte,
+	responseCode int, contentType string, response interface{}) http.Header {
+	body := bytes.NewBuffer(nil)
+	mw := multipart.NewWriter(body)
+	hdr := make(textproto.MIMEHeader)
+	hdr.Set("Content-Disposition", fmt.Sprintf("form-data; name=%q", fileName))
+	hdr.Set("Content-Type", contentType)
+	hdr.Set("Content-Length", strconv.FormatInt(int64(len(data)), 10))
+	part, err := mw.CreatePart(hdr)
+	if err != nil {
+		t.Error(err)
+	}
+	_, err = io.Copy(part, bytes.NewReader(data))
+	if err != nil {
+		t.Error(err)
+	}
+	err = mw.Close()
+	if err != nil {
+		t.Error(err)
+	}
+
+	req, err := http.NewRequest(method, url, body)
+	if err != nil {
+		t.Error(err)
+	}
+	req.Header.Set("Content-Type", fmt.Sprintf("multipart/form-data; boundary=%q", mw.Boundary()))
+
+	res, err := client.Do(req)
+	if err != nil {
+		t.Error(err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode != responseCode {
+		t.Errorf("got response status %s, want %v %s", res.Status, responseCode, http.StatusText(responseCode))
+	}
+
+	got, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got = bytes.TrimSpace(got)
+
+	want, err := json.Marshal(response)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if !bytes.Equal(got, want) {
+		t.Errorf("got response %s, want %s", string(got), string(want))
+	}
+
+	return res.Header
+}
+func ResponseDirectCheckBinaryResponse(t *testing.T, client *http.Client, method, url string, body io.Reader, responseCode int,
+	response []byte, headers http.Header) http.Header {
+	t.Helper()
+
+	resp := request(t, client, method, url, body, responseCode, headers)
+	defer resp.Body.Close()
+
+	got, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got = bytes.TrimSpace(got)
+
+	if !bytes.Equal(got, response) {
+		t.Errorf("got response %s, want %s", string(got), string(response))
+	}
+	return resp.Header
+}
+
 func ResponseDirectSendHeadersAndReceiveHeaders(t *testing.T, client *http.Client, method, url string, body io.Reader, responseCode int,
 	response interface{}, headers http.Header) http.Header {
 	t.Helper()