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.

455 lines
13 KiB

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