filer_multipart.go 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779
  1. package s3api
  2. import (
  3. "cmp"
  4. "crypto/rand"
  5. "encoding/base64"
  6. "encoding/hex"
  7. "encoding/xml"
  8. "fmt"
  9. "math"
  10. "path/filepath"
  11. "slices"
  12. "sort"
  13. "strconv"
  14. "strings"
  15. "time"
  16. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  17. "github.com/seaweedfs/seaweedfs/weed/stats"
  18. "github.com/aws/aws-sdk-go/aws"
  19. "github.com/aws/aws-sdk-go/service/s3"
  20. "github.com/google/uuid"
  21. "github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
  22. "net/http"
  23. "github.com/seaweedfs/seaweedfs/weed/filer"
  24. "github.com/seaweedfs/seaweedfs/weed/glog"
  25. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  26. )
  27. const (
  28. multipartExt = ".part"
  29. multiPartMinSize = 5 * 1024 * 1024
  30. )
  31. type InitiateMultipartUploadResult struct {
  32. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
  33. s3.CreateMultipartUploadOutput
  34. }
  35. func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code s3err.ErrorCode) {
  36. glog.V(2).Infof("createMultipartUpload input %v", input)
  37. uploadIdString := s3a.generateUploadID(*input.Key)
  38. uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "")
  39. // Prepare error handling outside callback scope
  40. var encryptionError error
  41. if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
  42. if entry.Extended == nil {
  43. entry.Extended = make(map[string][]byte)
  44. }
  45. entry.Extended["key"] = []byte(*input.Key)
  46. // Set object owner for multipart upload
  47. amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
  48. if amzAccountId != "" {
  49. entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
  50. }
  51. for k, v := range input.Metadata {
  52. entry.Extended[k] = []byte(*v)
  53. }
  54. if input.ContentType != nil {
  55. entry.Attributes.Mime = *input.ContentType
  56. }
  57. // Prepare and apply encryption configuration within directory creation
  58. // This ensures encryption resources are only allocated if directory creation succeeds
  59. encryptionConfig, prepErr := s3a.prepareMultipartEncryptionConfig(r, uploadIdString)
  60. if prepErr != nil {
  61. encryptionError = prepErr
  62. return // Exit callback, letting mkdir handle the error
  63. }
  64. s3a.applyMultipartEncryptionConfig(entry, encryptionConfig)
  65. // Extract and store object lock metadata from request headers
  66. // This ensures object lock settings from create_multipart_upload are preserved
  67. if err := s3a.extractObjectLockMetadataFromRequest(r, entry); err != nil {
  68. glog.Errorf("createMultipartUpload: failed to extract object lock metadata: %v", err)
  69. // Don't fail the upload - this matches AWS behavior for invalid metadata
  70. }
  71. }); err != nil {
  72. _, errorCode := handleMultipartInternalError("create multipart upload directory", err)
  73. return nil, errorCode
  74. }
  75. // Check for encryption configuration errors that occurred within the callback
  76. if encryptionError != nil {
  77. _, errorCode := handleMultipartInternalError("prepare encryption configuration", encryptionError)
  78. return nil, errorCode
  79. }
  80. output = &InitiateMultipartUploadResult{
  81. CreateMultipartUploadOutput: s3.CreateMultipartUploadOutput{
  82. Bucket: input.Bucket,
  83. Key: objectKey(input.Key),
  84. UploadId: aws.String(uploadIdString),
  85. },
  86. }
  87. return
  88. }
  89. type CompleteMultipartUploadResult struct {
  90. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
  91. Location *string `xml:"Location,omitempty"`
  92. Bucket *string `xml:"Bucket,omitempty"`
  93. Key *string `xml:"Key,omitempty"`
  94. ETag *string `xml:"ETag,omitempty"`
  95. // VersionId is NOT included in XML body - it should only be in x-amz-version-id HTTP header
  96. // Store the VersionId internally for setting HTTP header, but don't marshal to XML
  97. VersionId *string `xml:"-"`
  98. }
  99. func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
  100. glog.V(2).Infof("completeMultipartUpload input %v", input)
  101. if len(parts.Parts) == 0 {
  102. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
  103. return nil, s3err.ErrNoSuchUpload
  104. }
  105. completedPartNumbers := []int{}
  106. completedPartMap := make(map[int][]string)
  107. maxPartNo := 1
  108. for _, part := range parts.Parts {
  109. if _, ok := completedPartMap[part.PartNumber]; !ok {
  110. completedPartNumbers = append(completedPartNumbers, part.PartNumber)
  111. }
  112. completedPartMap[part.PartNumber] = append(completedPartMap[part.PartNumber], part.ETag)
  113. maxPartNo = maxInt(maxPartNo, part.PartNumber)
  114. }
  115. sort.Ints(completedPartNumbers)
  116. uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
  117. entries, _, err := s3a.list(uploadDirectory, "", "", false, 0)
  118. if err != nil {
  119. glog.Errorf("completeMultipartUpload %s %s error: %v, entries:%d", *input.Bucket, *input.UploadId, err, len(entries))
  120. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
  121. return nil, s3err.ErrNoSuchUpload
  122. }
  123. if len(entries) == 0 {
  124. entryName, dirName := s3a.getEntryNameAndDir(input)
  125. if entry, _ := s3a.getEntry(dirName, entryName); entry != nil && entry.Extended != nil {
  126. if uploadId, ok := entry.Extended[s3_constants.SeaweedFSUploadId]; ok && *input.UploadId == string(uploadId) {
  127. return &CompleteMultipartUploadResult{
  128. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
  129. Bucket: input.Bucket,
  130. ETag: aws.String("\"" + filer.ETagChunks(entry.GetChunks()) + "\""),
  131. Key: objectKey(input.Key),
  132. }, s3err.ErrNone
  133. }
  134. }
  135. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
  136. return nil, s3err.ErrNoSuchUpload
  137. }
  138. pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
  139. if err != nil {
  140. glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
  141. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
  142. return nil, s3err.ErrNoSuchUpload
  143. }
  144. deleteEntries := []*filer_pb.Entry{}
  145. partEntries := make(map[int][]*filer_pb.Entry, len(entries))
  146. entityTooSmall := false
  147. for _, entry := range entries {
  148. foundEntry := false
  149. glog.V(4).Infof("completeMultipartUpload part entries %s", entry.Name)
  150. if entry.IsDirectory || !strings.HasSuffix(entry.Name, multipartExt) {
  151. continue
  152. }
  153. partNumber, err := parsePartNumber(entry.Name)
  154. if err != nil {
  155. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNumber).Inc()
  156. glog.Errorf("completeMultipartUpload failed to pasre partNumber %s:%s", entry.Name, err)
  157. continue
  158. }
  159. completedPartsByNumber, ok := completedPartMap[partNumber]
  160. if !ok {
  161. continue
  162. }
  163. for _, partETag := range completedPartsByNumber {
  164. partETag = strings.Trim(partETag, `"`)
  165. entryETag := hex.EncodeToString(entry.Attributes.GetMd5())
  166. if partETag != "" && len(partETag) == 32 && entryETag != "" {
  167. if entryETag != partETag {
  168. glog.Errorf("completeMultipartUpload %s ETag mismatch chunk: %s part: %s", entry.Name, entryETag, partETag)
  169. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagMismatch).Inc()
  170. continue
  171. }
  172. } else {
  173. glog.Warningf("invalid complete etag %s, partEtag %s", partETag, entryETag)
  174. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagInvalid).Inc()
  175. }
  176. if len(entry.Chunks) == 0 && partNumber != maxPartNo {
  177. glog.Warningf("completeMultipartUpload %s empty chunks", entry.Name)
  178. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEmpty).Inc()
  179. continue
  180. }
  181. //there maybe multi same part, because of client retry
  182. partEntries[partNumber] = append(partEntries[partNumber], entry)
  183. foundEntry = true
  184. }
  185. if foundEntry {
  186. if len(completedPartNumbers) > 1 && partNumber != completedPartNumbers[len(completedPartNumbers)-1] &&
  187. entry.Attributes.FileSize < multiPartMinSize {
  188. glog.Warningf("completeMultipartUpload %s part file size less 5mb", entry.Name)
  189. entityTooSmall = true
  190. }
  191. } else {
  192. deleteEntries = append(deleteEntries, entry)
  193. }
  194. }
  195. if entityTooSmall {
  196. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompleteEntityTooSmall).Inc()
  197. return nil, s3err.ErrEntityTooSmall
  198. }
  199. mime := pentry.Attributes.Mime
  200. var finalParts []*filer_pb.FileChunk
  201. var offset int64
  202. for _, partNumber := range completedPartNumbers {
  203. partEntriesByNumber, ok := partEntries[partNumber]
  204. if !ok {
  205. glog.Errorf("part %d has no entry", partNumber)
  206. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNotFound).Inc()
  207. return nil, s3err.ErrInvalidPart
  208. }
  209. found := false
  210. if len(partEntriesByNumber) > 1 {
  211. slices.SortFunc(partEntriesByNumber, func(a, b *filer_pb.Entry) int {
  212. return cmp.Compare(b.Chunks[0].ModifiedTsNs, a.Chunks[0].ModifiedTsNs)
  213. })
  214. }
  215. for _, entry := range partEntriesByNumber {
  216. if found {
  217. deleteEntries = append(deleteEntries, entry)
  218. stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEntryMismatch).Inc()
  219. continue
  220. }
  221. // Track within-part offset for SSE-KMS IV calculation
  222. var withinPartOffset int64 = 0
  223. for _, chunk := range entry.GetChunks() {
  224. // Update SSE metadata with correct within-part offset (unified approach for KMS and SSE-C)
  225. sseKmsMetadata := chunk.SseMetadata
  226. if chunk.SseType == filer_pb.SSEType_SSE_KMS && len(chunk.SseMetadata) > 0 {
  227. // Deserialize, update offset, and re-serialize SSE-KMS metadata
  228. if kmsKey, err := DeserializeSSEKMSMetadata(chunk.SseMetadata); err == nil {
  229. kmsKey.ChunkOffset = withinPartOffset
  230. if updatedMetadata, serErr := SerializeSSEKMSMetadata(kmsKey); serErr == nil {
  231. sseKmsMetadata = updatedMetadata
  232. glog.V(4).Infof("Updated SSE-KMS metadata for chunk in part %d: withinPartOffset=%d", partNumber, withinPartOffset)
  233. }
  234. }
  235. } else if chunk.SseType == filer_pb.SSEType_SSE_C {
  236. // For SSE-C chunks, create per-chunk metadata using the part's IV
  237. if ivData, exists := entry.Extended[s3_constants.SeaweedFSSSEIV]; exists {
  238. // Get keyMD5 from entry metadata if available
  239. var keyMD5 string
  240. if keyMD5Data, keyExists := entry.Extended[s3_constants.AmzServerSideEncryptionCustomerKeyMD5]; keyExists {
  241. keyMD5 = string(keyMD5Data)
  242. }
  243. // Create SSE-C metadata with the part's IV and this chunk's within-part offset
  244. if ssecMetadata, serErr := SerializeSSECMetadata(ivData, keyMD5, withinPartOffset); serErr == nil {
  245. sseKmsMetadata = ssecMetadata // Reuse the same field for unified handling
  246. glog.V(4).Infof("Created SSE-C metadata for chunk in part %d: withinPartOffset=%d", partNumber, withinPartOffset)
  247. } else {
  248. glog.Errorf("Failed to serialize SSE-C metadata for chunk in part %d: %v", partNumber, serErr)
  249. }
  250. } else {
  251. glog.Errorf("SSE-C chunk in part %d missing IV in entry metadata", partNumber)
  252. }
  253. }
  254. p := &filer_pb.FileChunk{
  255. FileId: chunk.GetFileIdString(),
  256. Offset: offset,
  257. Size: chunk.Size,
  258. ModifiedTsNs: chunk.ModifiedTsNs,
  259. CipherKey: chunk.CipherKey,
  260. ETag: chunk.ETag,
  261. IsCompressed: chunk.IsCompressed,
  262. // Preserve SSE metadata with updated within-part offset
  263. SseType: chunk.SseType,
  264. SseMetadata: sseKmsMetadata,
  265. }
  266. finalParts = append(finalParts, p)
  267. offset += int64(chunk.Size)
  268. withinPartOffset += int64(chunk.Size)
  269. }
  270. found = true
  271. }
  272. }
  273. entryName, dirName := s3a.getEntryNameAndDir(input)
  274. // Check if versioning is configured for this bucket BEFORE creating any files
  275. versioningState, vErr := s3a.getVersioningState(*input.Bucket)
  276. if vErr == nil && versioningState == s3_constants.VersioningEnabled {
  277. // For versioned buckets, create a version and return the version ID
  278. versionId := generateVersionId()
  279. versionFileName := s3a.getVersionFileName(versionId)
  280. versionDir := dirName + "/" + entryName + ".versions"
  281. // Move the completed object to the versions directory
  282. err = s3a.mkFile(versionDir, versionFileName, finalParts, func(versionEntry *filer_pb.Entry) {
  283. if versionEntry.Extended == nil {
  284. versionEntry.Extended = make(map[string][]byte)
  285. }
  286. versionEntry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionId)
  287. versionEntry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
  288. // Set object owner for versioned multipart objects
  289. amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
  290. if amzAccountId != "" {
  291. versionEntry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
  292. }
  293. for k, v := range pentry.Extended {
  294. if k != "key" {
  295. versionEntry.Extended[k] = v
  296. }
  297. }
  298. // Preserve SSE-KMS metadata from the first part (if any)
  299. // SSE-KMS metadata is stored in individual parts, not the upload directory
  300. if len(completedPartNumbers) > 0 && len(partEntries[completedPartNumbers[0]]) > 0 {
  301. firstPartEntry := partEntries[completedPartNumbers[0]][0]
  302. if firstPartEntry.Extended != nil {
  303. // Copy SSE-KMS metadata from the first part
  304. if kmsMetadata, exists := firstPartEntry.Extended[s3_constants.SeaweedFSSSEKMSKey]; exists {
  305. versionEntry.Extended[s3_constants.SeaweedFSSSEKMSKey] = kmsMetadata
  306. glog.V(3).Infof("completeMultipartUpload: preserved SSE-KMS metadata from first part (versioned)")
  307. }
  308. }
  309. }
  310. if pentry.Attributes.Mime != "" {
  311. versionEntry.Attributes.Mime = pentry.Attributes.Mime
  312. } else if mime != "" {
  313. versionEntry.Attributes.Mime = mime
  314. }
  315. versionEntry.Attributes.FileSize = uint64(offset)
  316. })
  317. if err != nil {
  318. glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err)
  319. return nil, s3err.ErrInternalError
  320. }
  321. // Update the .versions directory metadata to indicate this is the latest version
  322. err = s3a.updateLatestVersionInDirectory(*input.Bucket, *input.Key, versionId, versionFileName)
  323. if err != nil {
  324. glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err)
  325. return nil, s3err.ErrInternalError
  326. }
  327. // For versioned buckets, don't create a main object file - all content is stored in .versions directory
  328. // The latest version information is tracked in the .versions directory metadata
  329. output = &CompleteMultipartUploadResult{
  330. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
  331. Bucket: input.Bucket,
  332. ETag: aws.String("\"" + filer.ETagChunks(finalParts) + "\""),
  333. Key: objectKey(input.Key),
  334. VersionId: aws.String(versionId),
  335. }
  336. } else if vErr == nil && versioningState == s3_constants.VersioningSuspended {
  337. // For suspended versioning, add "null" version ID metadata and return "null" version ID
  338. err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
  339. if entry.Extended == nil {
  340. entry.Extended = make(map[string][]byte)
  341. }
  342. entry.Extended[s3_constants.ExtVersionIdKey] = []byte("null")
  343. // Set object owner for suspended versioning multipart objects
  344. amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
  345. if amzAccountId != "" {
  346. entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
  347. }
  348. for k, v := range pentry.Extended {
  349. if k != "key" {
  350. entry.Extended[k] = v
  351. }
  352. }
  353. // Preserve SSE-KMS metadata from the first part (if any)
  354. // SSE-KMS metadata is stored in individual parts, not the upload directory
  355. if len(completedPartNumbers) > 0 && len(partEntries[completedPartNumbers[0]]) > 0 {
  356. firstPartEntry := partEntries[completedPartNumbers[0]][0]
  357. if firstPartEntry.Extended != nil {
  358. // Copy SSE-KMS metadata from the first part
  359. if kmsMetadata, exists := firstPartEntry.Extended[s3_constants.SeaweedFSSSEKMSKey]; exists {
  360. entry.Extended[s3_constants.SeaweedFSSSEKMSKey] = kmsMetadata
  361. glog.V(3).Infof("completeMultipartUpload: preserved SSE-KMS metadata from first part (suspended versioning)")
  362. }
  363. }
  364. }
  365. if pentry.Attributes.Mime != "" {
  366. entry.Attributes.Mime = pentry.Attributes.Mime
  367. } else if mime != "" {
  368. entry.Attributes.Mime = mime
  369. }
  370. entry.Attributes.FileSize = uint64(offset)
  371. })
  372. if err != nil {
  373. glog.Errorf("completeMultipartUpload: failed to create suspended versioning object: %v", err)
  374. return nil, s3err.ErrInternalError
  375. }
  376. // Note: Suspended versioning should NOT return VersionId field according to AWS S3 spec
  377. output = &CompleteMultipartUploadResult{
  378. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
  379. Bucket: input.Bucket,
  380. ETag: aws.String("\"" + filer.ETagChunks(finalParts) + "\""),
  381. Key: objectKey(input.Key),
  382. // VersionId field intentionally omitted for suspended versioning
  383. }
  384. } else {
  385. // For non-versioned buckets, create main object file
  386. err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
  387. if entry.Extended == nil {
  388. entry.Extended = make(map[string][]byte)
  389. }
  390. entry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
  391. // Set object owner for non-versioned multipart objects
  392. amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
  393. if amzAccountId != "" {
  394. entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
  395. }
  396. for k, v := range pentry.Extended {
  397. if k != "key" {
  398. entry.Extended[k] = v
  399. }
  400. }
  401. // Preserve SSE-KMS metadata from the first part (if any)
  402. // SSE-KMS metadata is stored in individual parts, not the upload directory
  403. if len(completedPartNumbers) > 0 && len(partEntries[completedPartNumbers[0]]) > 0 {
  404. firstPartEntry := partEntries[completedPartNumbers[0]][0]
  405. if firstPartEntry.Extended != nil {
  406. // Copy SSE-KMS metadata from the first part
  407. if kmsMetadata, exists := firstPartEntry.Extended[s3_constants.SeaweedFSSSEKMSKey]; exists {
  408. entry.Extended[s3_constants.SeaweedFSSSEKMSKey] = kmsMetadata
  409. glog.V(3).Infof("completeMultipartUpload: preserved SSE-KMS metadata from first part")
  410. }
  411. }
  412. }
  413. if pentry.Attributes.Mime != "" {
  414. entry.Attributes.Mime = pentry.Attributes.Mime
  415. } else if mime != "" {
  416. entry.Attributes.Mime = mime
  417. }
  418. entry.Attributes.FileSize = uint64(offset)
  419. })
  420. if err != nil {
  421. glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
  422. return nil, s3err.ErrInternalError
  423. }
  424. // For non-versioned buckets, return response without VersionId
  425. output = &CompleteMultipartUploadResult{
  426. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
  427. Bucket: input.Bucket,
  428. ETag: aws.String("\"" + filer.ETagChunks(finalParts) + "\""),
  429. Key: objectKey(input.Key),
  430. }
  431. }
  432. for _, deleteEntry := range deleteEntries {
  433. //delete unused part data
  434. glog.Infof("completeMultipartUpload cleanup %s upload %s unused %s", *input.Bucket, *input.UploadId, deleteEntry.Name)
  435. if err = s3a.rm(uploadDirectory, deleteEntry.Name, true, true); err != nil {
  436. glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err)
  437. }
  438. }
  439. if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
  440. glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
  441. }
  442. return
  443. }
  444. func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) {
  445. entryName := filepath.Base(*input.Key)
  446. dirName := filepath.ToSlash(filepath.Dir(*input.Key))
  447. if dirName == "." {
  448. dirName = ""
  449. }
  450. if strings.HasPrefix(dirName, "/") {
  451. dirName = dirName[1:]
  452. }
  453. dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName)
  454. // remove suffix '/'
  455. if strings.HasSuffix(dirName, "/") {
  456. dirName = dirName[:len(dirName)-1]
  457. }
  458. return entryName, dirName
  459. }
  460. func parsePartNumber(fileName string) (int, error) {
  461. var partNumberString string
  462. index := strings.Index(fileName, "_")
  463. if index != -1 {
  464. partNumberString = fileName[:index]
  465. } else {
  466. partNumberString = fileName[:len(fileName)-len(multipartExt)]
  467. }
  468. return strconv.Atoi(partNumberString)
  469. }
  470. func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) {
  471. glog.V(2).Infof("abortMultipartUpload input %v", input)
  472. exists, err := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
  473. if err != nil {
  474. glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
  475. return nil, s3err.ErrNoSuchUpload
  476. }
  477. if exists {
  478. err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
  479. }
  480. if err != nil {
  481. glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
  482. return nil, s3err.ErrInternalError
  483. }
  484. return &s3.AbortMultipartUploadOutput{}, s3err.ErrNone
  485. }
  486. type ListMultipartUploadsResult struct {
  487. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
  488. // copied from s3.ListMultipartUploadsOutput, the Uploads is not converting to <Upload></Upload>
  489. Bucket *string `type:"string"`
  490. Delimiter *string `type:"string"`
  491. EncodingType *string `type:"string" enum:"EncodingType"`
  492. IsTruncated *bool `type:"boolean"`
  493. KeyMarker *string `type:"string"`
  494. MaxUploads *int64 `type:"integer"`
  495. NextKeyMarker *string `type:"string"`
  496. NextUploadIdMarker *string `type:"string"`
  497. Prefix *string `type:"string"`
  498. UploadIdMarker *string `type:"string"`
  499. Upload []*s3.MultipartUpload `locationName:"Upload" type:"list" flattened:"true"`
  500. }
  501. func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code s3err.ErrorCode) {
  502. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListMultipartUploads.html
  503. glog.V(2).Infof("listMultipartUploads input %v", input)
  504. output = &ListMultipartUploadsResult{
  505. Bucket: input.Bucket,
  506. Delimiter: input.Delimiter,
  507. EncodingType: input.EncodingType,
  508. KeyMarker: input.KeyMarker,
  509. MaxUploads: input.MaxUploads,
  510. Prefix: input.Prefix,
  511. IsTruncated: aws.Bool(false),
  512. }
  513. entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", *input.UploadIdMarker, false, math.MaxInt32)
  514. if err != nil {
  515. glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
  516. return
  517. }
  518. uploadsCount := int64(0)
  519. for _, entry := range entries {
  520. if entry.Extended != nil {
  521. key := string(entry.Extended["key"])
  522. if *input.KeyMarker != "" && *input.KeyMarker != key {
  523. continue
  524. }
  525. if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
  526. continue
  527. }
  528. output.Upload = append(output.Upload, &s3.MultipartUpload{
  529. Key: objectKey(aws.String(key)),
  530. UploadId: aws.String(entry.Name),
  531. })
  532. uploadsCount += 1
  533. }
  534. if uploadsCount >= *input.MaxUploads {
  535. output.IsTruncated = aws.Bool(true)
  536. output.NextUploadIdMarker = aws.String(entry.Name)
  537. break
  538. }
  539. }
  540. return
  541. }
  542. type ListPartsResult struct {
  543. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
  544. // copied from s3.ListPartsOutput, the Parts is not converting to <Part></Part>
  545. Bucket *string `type:"string"`
  546. IsTruncated *bool `type:"boolean"`
  547. Key *string `min:"1" type:"string"`
  548. MaxParts *int64 `type:"integer"`
  549. NextPartNumberMarker *int64 `type:"integer"`
  550. PartNumberMarker *int64 `type:"integer"`
  551. Part []*s3.Part `locationName:"Part" type:"list" flattened:"true"`
  552. StorageClass *string `type:"string" enum:"StorageClass"`
  553. UploadId *string `type:"string"`
  554. }
  555. func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListPartsResult, code s3err.ErrorCode) {
  556. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListParts.html
  557. glog.V(2).Infof("listObjectParts input %v", input)
  558. output = &ListPartsResult{
  559. Bucket: input.Bucket,
  560. Key: objectKey(input.Key),
  561. UploadId: input.UploadId,
  562. MaxParts: input.MaxParts, // the maximum number of parts to return.
  563. PartNumberMarker: input.PartNumberMarker, // the part number starts after this, exclusive
  564. StorageClass: aws.String("STANDARD"),
  565. }
  566. entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d%s", *input.PartNumberMarker, multipartExt), false, uint32(*input.MaxParts))
  567. if err != nil {
  568. glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
  569. return nil, s3err.ErrNoSuchUpload
  570. }
  571. // Note: The upload directory is sort of a marker of the existence of an multipart upload request.
  572. // So can not just delete empty upload folders.
  573. output.IsTruncated = aws.Bool(!isLast)
  574. for _, entry := range entries {
  575. if strings.HasSuffix(entry.Name, multipartExt) && !entry.IsDirectory {
  576. partNumber, err := parsePartNumber(entry.Name)
  577. if err != nil {
  578. glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err)
  579. continue
  580. }
  581. output.Part = append(output.Part, &s3.Part{
  582. PartNumber: aws.Int64(int64(partNumber)),
  583. LastModified: aws.Time(time.Unix(entry.Attributes.Mtime, 0).UTC()),
  584. Size: aws.Int64(int64(filer.FileSize(entry))),
  585. ETag: aws.String("\"" + filer.ETag(entry) + "\""),
  586. })
  587. if !isLast {
  588. output.NextPartNumberMarker = aws.Int64(int64(partNumber))
  589. }
  590. }
  591. }
  592. return
  593. }
  594. // maxInt returns the maximum of two int values
  595. func maxInt(a, b int) int {
  596. if a > b {
  597. return a
  598. }
  599. return b
  600. }
  601. // MultipartEncryptionConfig holds pre-prepared encryption configuration to avoid error handling in callbacks
  602. type MultipartEncryptionConfig struct {
  603. // SSE-KMS configuration
  604. IsSSEKMS bool
  605. KMSKeyID string
  606. BucketKeyEnabled bool
  607. EncryptionContext string
  608. KMSBaseIVEncoded string
  609. // SSE-S3 configuration
  610. IsSSES3 bool
  611. S3BaseIVEncoded string
  612. S3KeyDataEncoded string
  613. }
  614. // prepareMultipartEncryptionConfig prepares encryption configuration with proper error handling
  615. // This eliminates the need for criticalError variable in callback functions
  616. func (s3a *S3ApiServer) prepareMultipartEncryptionConfig(r *http.Request, uploadIdString string) (*MultipartEncryptionConfig, error) {
  617. config := &MultipartEncryptionConfig{}
  618. // Prepare SSE-KMS configuration
  619. if IsSSEKMSRequest(r) {
  620. config.IsSSEKMS = true
  621. config.KMSKeyID = r.Header.Get(s3_constants.AmzServerSideEncryptionAwsKmsKeyId)
  622. config.BucketKeyEnabled = strings.ToLower(r.Header.Get(s3_constants.AmzServerSideEncryptionBucketKeyEnabled)) == "true"
  623. config.EncryptionContext = r.Header.Get(s3_constants.AmzServerSideEncryptionContext)
  624. // Generate and encode base IV with proper error handling
  625. baseIV := make([]byte, s3_constants.AESBlockSize)
  626. n, err := rand.Read(baseIV)
  627. if err != nil || n != len(baseIV) {
  628. return nil, fmt.Errorf("failed to generate secure IV for SSE-KMS multipart upload: %v (read %d/%d bytes)", err, n, len(baseIV))
  629. }
  630. config.KMSBaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
  631. glog.V(4).Infof("Generated base IV %x for SSE-KMS multipart upload %s", baseIV[:8], uploadIdString)
  632. }
  633. // Prepare SSE-S3 configuration
  634. if IsSSES3RequestInternal(r) {
  635. config.IsSSES3 = true
  636. // Generate and encode base IV with proper error handling
  637. baseIV := make([]byte, s3_constants.AESBlockSize)
  638. n, err := rand.Read(baseIV)
  639. if err != nil || n != len(baseIV) {
  640. return nil, fmt.Errorf("failed to generate secure IV for SSE-S3 multipart upload: %v (read %d/%d bytes)", err, n, len(baseIV))
  641. }
  642. config.S3BaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
  643. glog.V(4).Infof("Generated base IV %x for SSE-S3 multipart upload %s", baseIV[:8], uploadIdString)
  644. // Generate and serialize SSE-S3 key with proper error handling
  645. keyManager := GetSSES3KeyManager()
  646. sseS3Key, err := keyManager.GetOrCreateKey("")
  647. if err != nil {
  648. return nil, fmt.Errorf("failed to generate SSE-S3 key for multipart upload: %v", err)
  649. }
  650. keyData, serErr := SerializeSSES3Metadata(sseS3Key)
  651. if serErr != nil {
  652. return nil, fmt.Errorf("failed to serialize SSE-S3 metadata for multipart upload: %v", serErr)
  653. }
  654. config.S3KeyDataEncoded = base64.StdEncoding.EncodeToString(keyData)
  655. // Store key in manager for later retrieval
  656. keyManager.StoreKey(sseS3Key)
  657. glog.V(4).Infof("Stored SSE-S3 key %s for multipart upload %s", sseS3Key.KeyID, uploadIdString)
  658. }
  659. return config, nil
  660. }
  661. // applyMultipartEncryptionConfig applies pre-prepared encryption configuration to filer entry
  662. // This function is guaranteed not to fail since all error-prone operations were done during preparation
  663. func (s3a *S3ApiServer) applyMultipartEncryptionConfig(entry *filer_pb.Entry, config *MultipartEncryptionConfig) {
  664. // Apply SSE-KMS configuration
  665. if config.IsSSEKMS {
  666. entry.Extended[s3_constants.SeaweedFSSSEKMSKeyID] = []byte(config.KMSKeyID)
  667. if config.BucketKeyEnabled {
  668. entry.Extended[s3_constants.SeaweedFSSSEKMSBucketKeyEnabled] = []byte("true")
  669. }
  670. if config.EncryptionContext != "" {
  671. entry.Extended[s3_constants.SeaweedFSSSEKMSEncryptionContext] = []byte(config.EncryptionContext)
  672. }
  673. entry.Extended[s3_constants.SeaweedFSSSEKMSBaseIV] = []byte(config.KMSBaseIVEncoded)
  674. glog.V(3).Infof("applyMultipartEncryptionConfig: applied SSE-KMS settings with keyID %s", config.KMSKeyID)
  675. }
  676. // Apply SSE-S3 configuration
  677. if config.IsSSES3 {
  678. entry.Extended[s3_constants.SeaweedFSSSES3Encryption] = []byte(s3_constants.SSEAlgorithmAES256)
  679. entry.Extended[s3_constants.SeaweedFSSSES3BaseIV] = []byte(config.S3BaseIVEncoded)
  680. entry.Extended[s3_constants.SeaweedFSSSES3KeyData] = []byte(config.S3KeyDataEncoded)
  681. glog.V(3).Infof("applyMultipartEncryptionConfig: applied SSE-S3 settings")
  682. }
  683. }