You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

339 lines
9.7 KiB

4 years ago
3 years ago
4 years ago
6 years ago
3 years ago
3 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
3 years ago
3 years ago
3 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
4 years ago
3 years ago
3 years ago
  1. package s3api
  2. import (
  3. "encoding/xml"
  4. "crypto/sha1"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
  8. "github.com/chrislusf/seaweedfs/weed/s3api/s3err"
  9. weed_server "github.com/chrislusf/seaweedfs/weed/server"
  10. "io"
  11. "net/http"
  12. "net/url"
  13. "strconv"
  14. "strings"
  15. "github.com/aws/aws-sdk-go/aws"
  16. "github.com/aws/aws-sdk-go/service/s3"
  17. )
  18. const (
  19. maxObjectListSizeLimit = 10000 // Limit number of objects in a listObjectsResponse.
  20. maxUploadsList = 10000 // Limit number of uploads in a listUploadsResponse.
  21. maxPartsList = 10000 // Limit number of parts in a listPartsResponse.
  22. globalMaxPartID = 100000
  23. )
  24. // NewMultipartUploadHandler - New multipart upload.
  25. func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
  26. bucket, object := xhttp.GetBucketAndObject(r)
  27. createMultipartUploadInput := &s3.CreateMultipartUploadInput{
  28. Bucket: aws.String(bucket),
  29. Key: objectKey(aws.String(object)),
  30. Metadata: make(map[string]*string),
  31. }
  32. metadata := weed_server.SaveAmzMetaData(r, nil, false)
  33. for k, v := range metadata {
  34. createMultipartUploadInput.Metadata[k] = aws.String(string(v))
  35. }
  36. contentType := r.Header.Get("Content-Type")
  37. if contentType != "" {
  38. createMultipartUploadInput.ContentType = &contentType
  39. }
  40. response, errCode := s3a.createMultipartUpload(createMultipartUploadInput)
  41. glog.V(2).Info("NewMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
  42. if errCode != s3err.ErrNone {
  43. s3err.WriteErrorResponse(w, r, errCode)
  44. return
  45. }
  46. writeSuccessResponseXML(w, r, response)
  47. }
  48. // CompleteMultipartUploadHandler - Completes multipart upload.
  49. func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
  50. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html
  51. bucket, object := xhttp.GetBucketAndObject(r)
  52. parts := &CompleteMultipartUpload{}
  53. if err := xmlDecoder(r.Body, parts, r.ContentLength); err != nil {
  54. s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
  55. return
  56. }
  57. // Get upload id.
  58. uploadID, _, _, _ := getObjectResources(r.URL.Query())
  59. err := s3a.checkUploadId(object, uploadID)
  60. if err != nil {
  61. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
  62. return
  63. }
  64. response, errCode := s3a.completeMultipartUpload(&s3.CompleteMultipartUploadInput{
  65. Bucket: aws.String(bucket),
  66. Key: objectKey(aws.String(object)),
  67. UploadId: aws.String(uploadID),
  68. }, parts)
  69. glog.V(2).Info("CompleteMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)), errCode)
  70. if errCode != s3err.ErrNone {
  71. s3err.WriteErrorResponse(w, r, errCode)
  72. return
  73. }
  74. writeSuccessResponseXML(w, r, response)
  75. }
  76. // AbortMultipartUploadHandler - Aborts multipart upload.
  77. func (s3a *S3ApiServer) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
  78. bucket, object := xhttp.GetBucketAndObject(r)
  79. // Get upload id.
  80. uploadID, _, _, _ := getObjectResources(r.URL.Query())
  81. err := s3a.checkUploadId(object, uploadID)
  82. if err != nil {
  83. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
  84. return
  85. }
  86. response, errCode := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{
  87. Bucket: aws.String(bucket),
  88. Key: objectKey(aws.String(object)),
  89. UploadId: aws.String(uploadID),
  90. })
  91. if errCode != s3err.ErrNone {
  92. s3err.WriteErrorResponse(w, r, errCode)
  93. return
  94. }
  95. glog.V(2).Info("AbortMultipartUploadHandler", string(s3err.EncodeXMLResponse(response)))
  96. writeSuccessResponseXML(w, r, response)
  97. }
  98. // ListMultipartUploadsHandler - Lists multipart uploads.
  99. func (s3a *S3ApiServer) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
  100. bucket, _ := xhttp.GetBucketAndObject(r)
  101. prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType := getBucketMultipartResources(r.URL.Query())
  102. if maxUploads < 0 {
  103. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxUploads)
  104. return
  105. }
  106. if keyMarker != "" {
  107. // Marker not common with prefix is not implemented.
  108. if !strings.HasPrefix(keyMarker, prefix) {
  109. s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
  110. return
  111. }
  112. }
  113. response, errCode := s3a.listMultipartUploads(&s3.ListMultipartUploadsInput{
  114. Bucket: aws.String(bucket),
  115. Delimiter: aws.String(delimiter),
  116. EncodingType: aws.String(encodingType),
  117. KeyMarker: aws.String(keyMarker),
  118. MaxUploads: aws.Int64(int64(maxUploads)),
  119. Prefix: aws.String(prefix),
  120. UploadIdMarker: aws.String(uploadIDMarker),
  121. })
  122. glog.V(2).Infof("ListMultipartUploadsHandler %s errCode=%d", string(s3err.EncodeXMLResponse(response)), errCode)
  123. if errCode != s3err.ErrNone {
  124. s3err.WriteErrorResponse(w, r, errCode)
  125. return
  126. }
  127. // TODO handle encodingType
  128. writeSuccessResponseXML(w, r, response)
  129. }
  130. // ListObjectPartsHandler - Lists object parts in a multipart upload.
  131. func (s3a *S3ApiServer) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
  132. bucket, object := xhttp.GetBucketAndObject(r)
  133. uploadID, partNumberMarker, maxParts, _ := getObjectResources(r.URL.Query())
  134. if partNumberMarker < 0 {
  135. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPartNumberMarker)
  136. return
  137. }
  138. if maxParts < 0 {
  139. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
  140. return
  141. }
  142. err := s3a.checkUploadId(object, uploadID)
  143. if err != nil {
  144. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
  145. return
  146. }
  147. response, errCode := s3a.listObjectParts(&s3.ListPartsInput{
  148. Bucket: aws.String(bucket),
  149. Key: objectKey(aws.String(object)),
  150. MaxParts: aws.Int64(int64(maxParts)),
  151. PartNumberMarker: aws.Int64(int64(partNumberMarker)),
  152. UploadId: aws.String(uploadID),
  153. })
  154. if errCode != s3err.ErrNone {
  155. s3err.WriteErrorResponse(w, r, errCode)
  156. return
  157. }
  158. glog.V(2).Infof("ListObjectPartsHandler %s count=%d", string(s3err.EncodeXMLResponse(response)), len(response.Part))
  159. writeSuccessResponseXML(w, r, response)
  160. }
  161. // PutObjectPartHandler - Put an object part in a multipart upload.
  162. func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
  163. bucket, object := xhttp.GetBucketAndObject(r)
  164. uploadID := r.URL.Query().Get("uploadId")
  165. err := s3a.checkUploadId(object, uploadID)
  166. if err != nil {
  167. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
  168. return
  169. }
  170. partIDString := r.URL.Query().Get("partNumber")
  171. partID, err := strconv.Atoi(partIDString)
  172. if err != nil {
  173. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
  174. return
  175. }
  176. if partID > globalMaxPartID {
  177. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
  178. return
  179. }
  180. dataReader := r.Body
  181. if s3a.iam.isEnabled() {
  182. rAuthType := getRequestAuthType(r)
  183. var s3ErrCode s3err.ErrorCode
  184. switch rAuthType {
  185. case authTypeStreamingSigned:
  186. dataReader, s3ErrCode = s3a.iam.newSignV4ChunkedReader(r)
  187. case authTypeSignedV2, authTypePresignedV2:
  188. _, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
  189. case authTypePresigned, authTypeSigned:
  190. _, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
  191. }
  192. if s3ErrCode != s3err.ErrNone {
  193. s3err.WriteErrorResponse(w, r, s3ErrCode)
  194. return
  195. }
  196. }
  197. defer dataReader.Close()
  198. glog.V(2).Infof("PutObjectPartHandler %s %s %04d", bucket, uploadID, partID)
  199. uploadUrl := fmt.Sprintf("http://%s%s/%s/%04d.part?collection=%s",
  200. s3a.option.Filer.ToHttpAddress(), s3a.genUploadsFolder(bucket), uploadID, partID, bucket)
  201. if partID == 1 && r.Header.Get("Content-Type") == "" {
  202. dataReader = mimeDetect(r, dataReader)
  203. }
  204. etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
  205. if errCode != s3err.ErrNone {
  206. s3err.WriteErrorResponse(w, r, errCode)
  207. return
  208. }
  209. setEtag(w, etag)
  210. writeSuccessResponseEmpty(w, r)
  211. }
  212. func (s3a *S3ApiServer) genUploadsFolder(bucket string) string {
  213. return fmt.Sprintf("%s/%s/.uploads", s3a.option.BucketsPath, bucket)
  214. }
  215. // Generate uploadID hash string from object
  216. func (s3a *S3ApiServer) generateUploadID(object string) string {
  217. if strings.HasPrefix(object, "/") {
  218. object = object[1:]
  219. }
  220. h := sha1.New()
  221. h.Write([]byte(object))
  222. return fmt.Sprintf("%x", h.Sum(nil))
  223. }
  224. //Check object name and uploadID when processing multipart uploading
  225. func (s3a *S3ApiServer) checkUploadId(object string, id string) error {
  226. hash := s3a.generateUploadID(object)
  227. if hash != id {
  228. glog.Errorf("object %s and uploadID %s are not matched", object, id)
  229. return fmt.Errorf("object %s and uploadID %s are not matched", object, id)
  230. }
  231. return nil
  232. }
  233. // Parse bucket url queries for ?uploads
  234. func getBucketMultipartResources(values url.Values) (prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int, encodingType string) {
  235. prefix = values.Get("prefix")
  236. keyMarker = values.Get("key-marker")
  237. uploadIDMarker = values.Get("upload-id-marker")
  238. delimiter = values.Get("delimiter")
  239. if values.Get("max-uploads") != "" {
  240. maxUploads, _ = strconv.Atoi(values.Get("max-uploads"))
  241. } else {
  242. maxUploads = maxUploadsList
  243. }
  244. encodingType = values.Get("encoding-type")
  245. return
  246. }
  247. // Parse object url queries
  248. func getObjectResources(values url.Values) (uploadID string, partNumberMarker, maxParts int, encodingType string) {
  249. uploadID = values.Get("uploadId")
  250. partNumberMarker, _ = strconv.Atoi(values.Get("part-number-marker"))
  251. if values.Get("max-parts") != "" {
  252. maxParts, _ = strconv.Atoi(values.Get("max-parts"))
  253. } else {
  254. maxParts = maxPartsList
  255. }
  256. encodingType = values.Get("encoding-type")
  257. return
  258. }
  259. func xmlDecoder(body io.Reader, v interface{}, size int64) error {
  260. var lbody io.Reader
  261. if size > 0 {
  262. lbody = io.LimitReader(body, size)
  263. } else {
  264. lbody = body
  265. }
  266. d := xml.NewDecoder(lbody)
  267. d.CharsetReader = func(label string, input io.Reader) (io.Reader, error) {
  268. return input, nil
  269. }
  270. return d.Decode(v)
  271. }
  272. type CompleteMultipartUpload struct {
  273. Parts []CompletedPart `xml:"Part"`
  274. }
  275. type CompletedPart struct {
  276. ETag string
  277. PartNumber int
  278. }