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.

381 lines
9.9 KiB

6 years ago
7 years ago
7 years ago
7 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
6 years ago
4 years ago
6 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. package s3api
  2. import (
  3. "crypto/md5"
  4. "encoding/json"
  5. "encoding/xml"
  6. "fmt"
  7. "io"
  8. "io/ioutil"
  9. "net/http"
  10. "sort"
  11. "strings"
  12. "github.com/chrislusf/seaweedfs/weed/s3api/s3err"
  13. "github.com/gorilla/mux"
  14. "github.com/chrislusf/seaweedfs/weed/glog"
  15. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  16. weed_server "github.com/chrislusf/seaweedfs/weed/server"
  17. "github.com/chrislusf/seaweedfs/weed/util"
  18. )
  19. var (
  20. client *http.Client
  21. )
  22. func init() {
  23. client = &http.Client{Transport: &http.Transport{
  24. MaxIdleConnsPerHost: 1024,
  25. }}
  26. }
  27. func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
  28. // http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
  29. bucket, object := getBucketAndObject(r)
  30. _, err := validateContentMd5(r.Header)
  31. if err != nil {
  32. writeErrorResponse(w, s3err.ErrInvalidDigest, r.URL)
  33. return
  34. }
  35. dataReader := r.Body
  36. if s3a.iam.isEnabled() {
  37. rAuthType := getRequestAuthType(r)
  38. var s3ErrCode s3err.ErrorCode
  39. switch rAuthType {
  40. case authTypeStreamingSigned:
  41. dataReader, s3ErrCode = s3a.iam.newSignV4ChunkedReader(r)
  42. case authTypeSignedV2, authTypePresignedV2:
  43. _, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
  44. case authTypePresigned, authTypeSigned:
  45. _, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
  46. }
  47. if s3ErrCode != s3err.ErrNone {
  48. writeErrorResponse(w, s3ErrCode, r.URL)
  49. return
  50. }
  51. }
  52. defer dataReader.Close()
  53. if strings.HasSuffix(object, "/") {
  54. if err := s3a.mkdir(s3a.option.BucketsPath, bucket+object, nil); err != nil {
  55. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  56. return
  57. }
  58. } else {
  59. uploadUrl := fmt.Sprintf("http://%s%s/%s%s", s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  60. etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
  61. if errCode != s3err.ErrNone {
  62. writeErrorResponse(w, errCode, r.URL)
  63. return
  64. }
  65. setEtag(w, etag)
  66. }
  67. writeSuccessResponseEmpty(w)
  68. }
  69. func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
  70. bucket, object := getBucketAndObject(r)
  71. if strings.HasSuffix(r.URL.Path, "/") {
  72. writeErrorResponse(w, s3err.ErrNotImplemented, r.URL)
  73. return
  74. }
  75. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  76. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  77. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  78. }
  79. func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
  80. bucket, object := getBucketAndObject(r)
  81. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  82. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  83. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  84. }
  85. func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
  86. bucket, object := getBucketAndObject(r)
  87. destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
  88. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  89. s3a.proxyToFiler(w, r, destUrl, func(proxyResponse *http.Response, w http.ResponseWriter) {
  90. for k, v := range proxyResponse.Header {
  91. w.Header()[k] = v
  92. }
  93. w.WriteHeader(http.StatusNoContent)
  94. })
  95. }
  96. // / ObjectIdentifier carries key name for the object to delete.
  97. type ObjectIdentifier struct {
  98. ObjectName string `xml:"Key"`
  99. }
  100. // DeleteObjectsRequest - xml carrying the object key names which needs to be deleted.
  101. type DeleteObjectsRequest struct {
  102. // Element to enable quiet mode for the request
  103. Quiet bool
  104. // List of objects to be deleted
  105. Objects []ObjectIdentifier `xml:"Object"`
  106. }
  107. // DeleteError structure.
  108. type DeleteError struct {
  109. Code string
  110. Message string
  111. Key string
  112. }
  113. // DeleteObjectsResponse container for multiple object deletes.
  114. type DeleteObjectsResponse struct {
  115. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ DeleteResult" json:"-"`
  116. // Collection of all deleted objects
  117. DeletedObjects []ObjectIdentifier `xml:"Deleted,omitempty"`
  118. // Collection of errors deleting certain objects.
  119. Errors []DeleteError `xml:"Error,omitempty"`
  120. }
  121. // DeleteMultipleObjectsHandler - Delete multiple objects
  122. func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
  123. bucket, _ := getBucketAndObject(r)
  124. deleteXMLBytes, err := ioutil.ReadAll(r.Body)
  125. if err != nil {
  126. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  127. return
  128. }
  129. deleteObjects := &DeleteObjectsRequest{}
  130. if err := xml.Unmarshal(deleteXMLBytes, deleteObjects); err != nil {
  131. writeErrorResponse(w, s3err.ErrMalformedXML, r.URL)
  132. return
  133. }
  134. var deletedObjects []ObjectIdentifier
  135. var deleteErrors []DeleteError
  136. directoriesWithDeletion := make(map[string]int)
  137. s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  138. // delete file entries
  139. for _, object := range deleteObjects.Objects {
  140. lastSeparator := strings.LastIndex(object.ObjectName, "/")
  141. parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, false
  142. if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
  143. entryName = object.ObjectName[lastSeparator+1:]
  144. parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
  145. }
  146. parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
  147. err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
  148. if err == nil {
  149. directoriesWithDeletion[parentDirectoryPath]++
  150. deletedObjects = append(deletedObjects, object)
  151. } else {
  152. delete(directoriesWithDeletion, parentDirectoryPath)
  153. deleteErrors = append(deleteErrors, DeleteError{
  154. Code: "",
  155. Message: err.Error(),
  156. Key: object.ObjectName,
  157. })
  158. }
  159. }
  160. // purge empty folders, only checking folders with deletions
  161. var allDirs []string
  162. for dir, _ := range directoriesWithDeletion {
  163. allDirs = append(allDirs, dir)
  164. }
  165. sort.Slice(allDirs, func(i, j int) bool {
  166. return len(allDirs[i]) > len(allDirs[j])
  167. })
  168. for _, dir := range allDirs {
  169. parentDir, dirName := util.FullPath(dir).DirAndName()
  170. if err := doDeleteEntry(client, parentDir, dirName, false, false); err != nil {
  171. glog.V(4).Infof("directory %s has %d deletion but still not empty: %v", dir, directoriesWithDeletion[dir], err)
  172. }
  173. }
  174. return nil
  175. })
  176. deleteResp := DeleteObjectsResponse{}
  177. if !deleteObjects.Quiet {
  178. deleteResp.DeletedObjects = deletedObjects
  179. }
  180. deleteResp.Errors = deleteErrors
  181. writeSuccessResponseXML(w, encodeResponse(deleteResp))
  182. }
  183. var passThroughHeaders = []string{
  184. "response-cache-control",
  185. "response-content-disposition",
  186. "response-content-encoding",
  187. "response-content-language",
  188. "response-content-type",
  189. "response-expires",
  190. }
  191. func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter)) {
  192. glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
  193. proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
  194. if err != nil {
  195. glog.Errorf("NewRequest %s: %v", destUrl, err)
  196. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  197. return
  198. }
  199. proxyReq.Header.Set("Host", s3a.option.Filer)
  200. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  201. for header, values := range r.Header {
  202. // handle s3 related headers
  203. passed := false
  204. for _, h := range passThroughHeaders {
  205. if strings.ToLower(header) == h && len(values) > 0 {
  206. proxyReq.Header.Add(header[len("response-"):], values[0])
  207. passed = true
  208. break
  209. }
  210. }
  211. if passed {
  212. continue
  213. }
  214. // handle other headers
  215. for _, value := range values {
  216. proxyReq.Header.Add(header, value)
  217. }
  218. }
  219. resp, postErr := client.Do(proxyReq)
  220. if postErr != nil {
  221. glog.Errorf("post to filer: %v", postErr)
  222. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  223. return
  224. }
  225. defer util.CloseResponse(resp)
  226. if (resp.ContentLength == -1 || resp.StatusCode == 404) && !strings.HasSuffix(destUrl, "/") {
  227. writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
  228. return
  229. }
  230. responseFn(resp, w)
  231. }
  232. func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) {
  233. for k, v := range proxyResponse.Header {
  234. w.Header()[k] = v
  235. }
  236. w.WriteHeader(proxyResponse.StatusCode)
  237. io.Copy(w, proxyResponse.Body)
  238. }
  239. func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader io.Reader) (etag string, code s3err.ErrorCode) {
  240. hash := md5.New()
  241. var body = io.TeeReader(dataReader, hash)
  242. proxyReq, err := http.NewRequest("PUT", uploadUrl, body)
  243. if err != nil {
  244. glog.Errorf("NewRequest %s: %v", uploadUrl, err)
  245. return "", s3err.ErrInternalError
  246. }
  247. proxyReq.Header.Set("Host", s3a.option.Filer)
  248. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  249. for header, values := range r.Header {
  250. for _, value := range values {
  251. proxyReq.Header.Add(header, value)
  252. }
  253. }
  254. resp, postErr := client.Do(proxyReq)
  255. if postErr != nil {
  256. glog.Errorf("post to filer: %v", postErr)
  257. return "", s3err.ErrInternalError
  258. }
  259. defer resp.Body.Close()
  260. etag = fmt.Sprintf("%x", hash.Sum(nil))
  261. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  262. if ra_err != nil {
  263. glog.Errorf("upload to filer response read %d: %v", resp.StatusCode, ra_err)
  264. return etag, s3err.ErrInternalError
  265. }
  266. var ret weed_server.FilerPostResult
  267. unmarshal_err := json.Unmarshal(resp_body, &ret)
  268. if unmarshal_err != nil {
  269. glog.Errorf("failing to read upload to %s : %v", uploadUrl, string(resp_body))
  270. return "", s3err.ErrInternalError
  271. }
  272. if ret.Error != "" {
  273. glog.Errorf("upload to filer error: %v", ret.Error)
  274. return "", filerErrorToS3Error(ret.Error)
  275. }
  276. return etag, s3err.ErrNone
  277. }
  278. func setEtag(w http.ResponseWriter, etag string) {
  279. if etag != "" {
  280. if strings.HasPrefix(etag, "\"") {
  281. w.Header().Set("ETag", etag)
  282. } else {
  283. w.Header().Set("ETag", "\""+etag+"\"")
  284. }
  285. }
  286. }
  287. func getBucketAndObject(r *http.Request) (bucket, object string) {
  288. vars := mux.Vars(r)
  289. bucket = vars["bucket"]
  290. object = vars["object"]
  291. if !strings.HasPrefix(object, "/") {
  292. object = "/" + object
  293. }
  294. return
  295. }
  296. func filerErrorToS3Error(errString string) s3err.ErrorCode {
  297. if strings.HasPrefix(errString, "existing ") && strings.HasSuffix(errString, "is a directory") {
  298. return s3err.ErrExistingObjectIsDirectory
  299. }
  300. return s3err.ErrInternalError
  301. }