magpie/s3.go
Maxfield Luke ddbd19f153 Initial commit on Forgejo
Fresh repository history for elektrine/magpie hosted at
https://git.elektrine.com/elektrine/magpie.
2026-07-29 04:56:34 -04:00

938 lines
28 KiB
Go

package main
import (
"context"
"crypto/md5"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/xml"
"errors"
"fmt"
"io"
"log"
"net/http"
"net/url"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
)
const (
maxXMLRequestBytes = 1 << 20
maxBatchDeleteItems = 1000
)
var errPayloadHashMismatch = errors.New("payload hash mismatch")
func (s *Server) s3Object(w http.ResponseWriter, r *http.Request) {
if s.s3BucketList(w, r) {
return
}
if s.s3BucketPost(w, r) {
return
}
bucket, key, ok := s.s3BucketKey(r.URL.Path)
if !ok {
s3Error(w, http.StatusNotFound, "NoSuchBucket", "bucket not found")
return
}
if !s.bucketAllowed(bucket) {
s3Error(w, http.StatusForbidden, "AccessDenied", "bucket is not allowed")
return
}
storeKey := bucket + "/" + key
if err := ValidateKey(storeKey); err != nil {
s3Error(w, http.StatusBadRequest, "InvalidObjectName", err.Error())
return
}
publicRead := (r.Method == http.MethodGet || r.Method == http.MethodHead) && s.publicReadAllowed(storeKey)
if !publicRead && !s.authorizedS3(r) {
s3Error(w, http.StatusForbidden, "SignatureDoesNotMatch", "request signature is invalid")
return
}
if !s.replicationAllowed(r) {
s3Error(w, http.StatusForbidden, "AccessDenied", "replication secret is invalid")
return
}
switch r.Method {
case http.MethodPost:
if _, ok := r.URL.Query()["uploads"]; ok {
s.s3CreateMultipartUpload(w, r, storeKey)
} else if r.URL.Query().Get("uploadId") != "" {
s.s3CompleteMultipartUpload(w, r, storeKey)
} else if _, ok := r.URL.Query()["delete"]; ok {
s.s3DeleteObjects(w, r, bucket)
} else {
s3Error(w, http.StatusMethodNotAllowed, "MethodNotAllowed", "method not allowed")
}
case http.MethodPut:
if r.URL.Query().Get("uploadId") != "" && r.URL.Query().Get("partNumber") != "" {
s.s3UploadPart(w, r, storeKey)
} else if r.Header.Get("X-Amz-Copy-Source") != "" {
s.s3CopyObject(w, r, storeKey)
} else {
s.s3Put(w, r, storeKey)
}
case http.MethodGet:
s.s3Get(w, r, storeKey, publicRead)
case http.MethodHead:
s.s3Head(w, r, storeKey, publicRead)
case http.MethodDelete:
if r.URL.Query().Get("uploadId") != "" {
s.s3AbortMultipartUpload(w, r, storeKey)
} else {
s.s3Delete(w, r, storeKey)
}
default:
w.Header().Set("Allow", "PUT, GET, HEAD, DELETE")
s3Error(w, http.StatusMethodNotAllowed, "MethodNotAllowed", "method not allowed")
}
}
func (s *Server) s3BucketPost(w http.ResponseWriter, r *http.Request) bool {
path := strings.Trim(r.URL.Path, "/")
if r.Method != http.MethodPost || path == "" || strings.Contains(path, "/") {
return false
}
if _, ok := r.URL.Query()["delete"]; !ok {
return false
}
if err := ValidateKey(path); err != nil {
s3Error(w, http.StatusBadRequest, "InvalidBucketName", err.Error())
return true
}
if !s.bucketAllowed(path) {
s3Error(w, http.StatusForbidden, "AccessDenied", "bucket is not allowed")
return true
}
if !s.authorizedS3(r) {
s3Error(w, http.StatusForbidden, "SignatureDoesNotMatch", "request signature is invalid")
return true
}
if !s.replicationAllowed(r) {
s3Error(w, http.StatusForbidden, "AccessDenied", "replication secret is invalid")
return true
}
s.s3DeleteObjects(w, r, path)
return true
}
func (s *Server) s3CreateMultipartUpload(w http.ResponseWriter, r *http.Request, key string) {
uploadID, err := randomID()
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "upload id failed")
return
}
dir := s.multipartDir(uploadID)
if err := os.MkdirAll(dir, 0o700); err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "upload init failed")
return
}
meta := multipartMeta{Key: key, ContentType: r.Header.Get("Content-Type")}
data, _ := xml.Marshal(meta)
if err := os.WriteFile(filepath.Join(dir, "meta.xml"), data, 0o600); err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "upload init failed")
return
}
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusOK)
_ = xml.NewEncoder(w).Encode(createMultipartUploadResult{UploadID: uploadID})
}
func (s *Server) s3UploadPart(w http.ResponseWriter, r *http.Request, key string) {
release, ok := s.acceptWrite(w, r, true)
if !ok {
return
}
defer release()
uploadID := r.URL.Query().Get("uploadId")
partNumber, err := strconv.Atoi(r.URL.Query().Get("partNumber"))
if err != nil || partNumber <= 0 || partNumber > 10_000 {
s3Error(w, http.StatusBadRequest, "InvalidArgument", "invalid part number")
return
}
meta, err := s.multipartMeta(uploadID)
if err != nil || meta.Key != key {
s3Error(w, http.StatusNotFound, "NoSuchUpload", "multipart upload not found")
return
}
dir := s.multipartDir(uploadID)
if !s.acceptMultipartPartSize(w, dir, r.ContentLength) {
return
}
partPath := filepath.Join(dir, fmt.Sprintf("part-%05d", partNumber))
body, verifyPayload, ok := s.s3StreamingBody(w, r)
if !ok {
return
}
tmp, err := os.CreateTemp(dir, ".part-*")
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "part upload failed")
return
}
hash := md5Hash()
_, copyErr := io.Copy(io.MultiWriter(tmp, hash), http.MaxBytesReader(w, body, s.cfg.MaxObjectSize))
closeErr := tmp.Close()
if copyErr != nil || closeErr != nil {
_ = os.Remove(tmp.Name())
if errors.Is(copyErr, errPayloadHashMismatch) {
s3Error(w, http.StatusForbidden, "XAmzContentSHA256Mismatch", "payload hash mismatch")
return
}
s3Error(w, http.StatusInternalServerError, "InternalError", "part upload failed")
return
}
if !verifyPayload() {
_ = os.Remove(tmp.Name())
s3Error(w, http.StatusForbidden, "XAmzContentSHA256Mismatch", "payload hash mismatch")
return
}
if err := os.Rename(tmp.Name(), partPath); err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "part upload failed")
return
}
etag := fmt.Sprintf("%x", hash.Sum(nil))
w.Header().Set("ETag", quoteETag(etag))
w.WriteHeader(http.StatusOK)
}
func (s *Server) acceptMultipartPartSize(w http.ResponseWriter, dir string, incomingSize int64) bool {
if incomingSize < 0 {
s3Error(w, http.StatusLengthRequired, "MissingContentLength", "content length is required")
return false
}
if s.cfg.MaxObjectSize <= 0 {
return true
}
existingSize := int64(0)
parts, _ := filepath.Glob(filepath.Join(dir, "part-*"))
for _, part := range parts {
info, err := os.Stat(part)
if err == nil && !info.IsDir() {
existingSize += info.Size()
}
}
if existingSize+incomingSize > s.cfg.MaxObjectSize {
s3Error(w, http.StatusRequestEntityTooLarge, "EntityTooLarge", "multipart upload exceeds object size limit")
return false
}
return true
}
func (s *Server) s3CompleteMultipartUpload(w http.ResponseWriter, r *http.Request, key string) {
uploadID := r.URL.Query().Get("uploadId")
meta, err := s.multipartMeta(uploadID)
if err != nil || meta.Key != key {
s3Error(w, http.StatusNotFound, "NoSuchUpload", "multipart upload not found")
return
}
var complete completeMultipartUpload
completeBody, ok := s.s3BoundedBody(w, r, maxXMLRequestBytes)
if !ok {
return
}
if err := xml.Unmarshal(completeBody, &complete); err != nil && len(completeBody) > 0 {
s3Error(w, http.StatusBadRequest, "MalformedXML", "invalid complete request")
return
}
parts := complete.Parts
if len(parts) == 0 {
entries, _ := filepath.Glob(filepath.Join(s.multipartDir(uploadID), "part-*"))
for _, entry := range entries {
base := filepath.Base(entry)
n, _ := strconv.Atoi(strings.TrimPrefix(base, "part-"))
parts = append(parts, completedPart{PartNumber: n})
}
}
if len(parts) > 10_000 {
s3Error(w, http.StatusBadRequest, "InvalidRequest", "too many parts")
return
}
sortParts(parts)
readers := make([]io.Reader, 0, len(parts))
files := make([]*os.File, 0, len(parts))
defer func() {
for _, file := range files {
_ = file.Close()
}
}()
var totalSize int64
for _, part := range parts {
if part.PartNumber <= 0 || part.PartNumber > 10_000 {
s3Error(w, http.StatusBadRequest, "InvalidPart", "invalid part number")
return
}
file, err := os.Open(filepath.Join(s.multipartDir(uploadID), fmt.Sprintf("part-%05d", part.PartNumber)))
if err != nil {
s3Error(w, http.StatusBadRequest, "InvalidPart", "part missing")
return
}
stat, err := file.Stat()
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "part stat failed")
return
}
totalSize += stat.Size()
if s.cfg.MaxObjectSize > 0 && totalSize > s.cfg.MaxObjectSize {
s3Error(w, http.StatusRequestEntityTooLarge, "EntityTooLarge", "object too large")
return
}
files = append(files, file)
readers = append(readers, file)
}
release, ok := s.acceptWriteSize(w, totalSize, true)
if !ok {
return
}
defer release()
info, err := s.store.Put(key, io.MultiReader(readers...), meta.ContentType)
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "complete failed")
return
}
_ = os.RemoveAll(s.multipartDir(uploadID))
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusOK)
_ = xml.NewEncoder(w).Encode(completeMultipartUploadResult{Key: strings.TrimPrefix(key, strings.SplitN(key, "/", 2)[0]+"/"), ETag: quoteETag(info.ETag)})
if r.Header.Get("X-Magpie-Replication") != "1" {
s.enqueueReplicationPut(key)
}
}
func (s *Server) s3AbortMultipartUpload(w http.ResponseWriter, r *http.Request, _ string) {
uploadID := r.URL.Query().Get("uploadId")
if !validUploadID(uploadID) {
s3Error(w, http.StatusNotFound, "NoSuchUpload", "multipart upload not found")
return
}
_ = os.RemoveAll(s.multipartDir(uploadID))
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) s3CopyObject(w http.ResponseWriter, r *http.Request, destinationKey string) {
if !signedHeaderIncludes(r, "x-amz-copy-source") {
s3Error(w, http.StatusForbidden, "AccessDenied", "copy source header must be signed")
return
}
sourceKey, err := copySourceKey(r.Header.Get("X-Amz-Copy-Source"))
if err != nil {
s3Error(w, http.StatusBadRequest, "InvalidArgument", "invalid copy source")
return
}
sourceBucket, _, _ := strings.Cut(sourceKey, "/")
if !s.bucketAllowed(sourceBucket) {
s3Error(w, http.StatusForbidden, "AccessDenied", "copy source bucket is not allowed")
return
}
if len(s.cfg.S3Keys) > 0 {
key, ok := s.requestAccessKey(r)
if !ok || !key.can("read") || !key.can("write") {
s3Error(w, http.StatusForbidden, "AccessDenied", "copy requires read and write permission")
return
}
}
obj, err := s.store.Open(sourceKey)
if err != nil {
if errors.Is(err, ErrNotFound) {
s3Error(w, http.StatusNotFound, "NoSuchKey", "copy source not found")
return
}
s3Error(w, http.StatusInternalServerError, "InternalError", "copy source failed")
return
}
defer obj.Close()
if s.cfg.MaxObjectSize > 0 && obj.Info.Size > s.cfg.MaxObjectSize {
s3Error(w, http.StatusRequestEntityTooLarge, "EntityTooLarge", "object too large")
return
}
release, ok := s.acceptWriteSize(w, obj.Info.Size, true)
if !ok {
return
}
defer release()
contentType := obj.Info.ContentType
if override := r.Header.Get("Content-Type"); override != "" && strings.EqualFold(r.Header.Get("X-Amz-Metadata-Directive"), "REPLACE") {
contentType = override
}
info, err := s.store.Put(destinationKey, obj, contentType)
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "copy failed")
return
}
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusOK)
_ = xml.NewEncoder(w).Encode(copyObjectResult{ETag: quoteETag(info.ETag), LastModified: info.ModTime.UTC().Format(time.RFC3339)})
if r.Header.Get("X-Magpie-Replication") != "1" {
s.enqueueReplicationPut(destinationKey)
}
}
func (s *Server) s3DeleteObjects(w http.ResponseWriter, r *http.Request, bucket string) {
var request deleteObjectsRequest
deleteBody, ok := s.s3BoundedBody(w, r, maxXMLRequestBytes)
if !ok {
return
}
if err := xml.Unmarshal(deleteBody, &request); err != nil {
s3Error(w, http.StatusBadRequest, "MalformedXML", "invalid delete request")
return
}
if len(request.Objects) > maxBatchDeleteItems {
s3Error(w, http.StatusBadRequest, "InvalidRequest", "too many objects in delete request")
return
}
result := deleteObjectsResult{Deleted: make([]deletedObject, 0, len(request.Objects))}
for _, object := range request.Objects {
key := strings.TrimPrefix(object.Key, "/")
storeKey := bucket + "/" + key
if err := ValidateKey(storeKey); err != nil {
result.Errors = append(result.Errors, deleteObjectError{Key: object.Key, Code: "InvalidObjectName", Message: err.Error()})
continue
}
if err := s.store.Delete(storeKey); err != nil && !errors.Is(err, ErrNotFound) {
result.Errors = append(result.Errors, deleteObjectError{Key: object.Key, Code: "InternalError", Message: "delete failed"})
continue
}
result.Deleted = append(result.Deleted, deletedObject{Key: object.Key})
if r.Header.Get("X-Magpie-Replication") != "1" {
s.enqueueReplicationDelete(storeKey)
}
}
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusOK)
_ = xml.NewEncoder(w).Encode(result)
}
func (s *Server) s3BucketList(w http.ResponseWriter, r *http.Request) bool {
if r.Method != http.MethodGet || strings.Trim(r.URL.Path, "/") == "" || strings.Contains(strings.Trim(r.URL.Path, "/"), "/") {
return false
}
if r.URL.RawQuery == "" && r.URL.Query().Get("prefix") == "" && r.URL.Query().Get("list-type") == "" {
return false
}
bucket := strings.Trim(r.URL.Path, "/")
if err := ValidateKey(bucket); err != nil {
s3Error(w, http.StatusBadRequest, "InvalidBucketName", err.Error())
return true
}
if !s.bucketAllowed(bucket) {
s3Error(w, http.StatusForbidden, "AccessDenied", "bucket is not allowed")
return true
}
if !s.authorizedS3(r) {
s3Error(w, http.StatusForbidden, "SignatureDoesNotMatch", "request signature is invalid")
return true
}
prefix := r.URL.Query().Get("prefix")
storePrefix := bucket + "/"
if prefix != "" {
storePrefix = bucket + "/" + strings.TrimPrefix(prefix, "/")
}
maxKeys := parseMaxKeys(r.URL.Query().Get("max-keys"))
continuationToken := r.URL.Query().Get("continuation-token")
after, err := decodeContinuationToken(continuationToken)
if err != nil {
s3Error(w, http.StatusBadRequest, "InvalidArgument", "invalid continuation token")
return true
}
page, err := s.store.ListPage(storePrefix, after, maxKeys)
if err != nil {
s3Error(w, http.StatusInternalServerError, "InternalError", "list failed")
return true
}
contents := make([]s3ListContent, 0, len(page.Objects))
for _, object := range page.Objects {
key := strings.TrimPrefix(object.Key, bucket+"/")
contents = append(contents, s3ListContent{
Key: key,
LastModified: object.ModTime.UTC().Format(time.RFC3339),
ETag: quoteETag(object.ETag),
Size: object.Size,
StorageClass: "STANDARD",
})
}
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusOK)
_ = xml.NewEncoder(w).Encode(s3ListBucketResult{
Name: bucket,
Prefix: prefix,
ContinuationToken: continuationToken,
NextContinuationToken: encodeContinuationToken(page.NextAfter),
KeyCount: len(contents),
MaxKeys: maxKeys,
IsTruncated: page.NextAfter != "",
Contents: contents,
})
return true
}
func encodeContinuationToken(key string) string {
if key == "" {
return ""
}
return base64.RawURLEncoding.EncodeToString([]byte(key))
}
func decodeContinuationToken(token string) (string, error) {
if token == "" {
return "", nil
}
decoded, err := base64.RawURLEncoding.DecodeString(token)
if err != nil {
return "", err
}
key := string(decoded)
if err := ValidateKey(key); err != nil {
return "", err
}
return key, nil
}
func copySourceKey(value string) (string, error) {
value = strings.TrimPrefix(value, "/")
if value == "" {
return "", fmt.Errorf("missing copy source")
}
if decoded, err := url.PathUnescape(value); err == nil {
value = decoded
}
if err := ValidateKey(value); err != nil {
return "", err
}
return value, nil
}
func (s *Server) s3BucketKey(path string) (string, string, bool) {
path = strings.TrimPrefix(path, "/")
parts := strings.SplitN(path, "/", 2)
if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
return "", "", false
}
if err := ValidateKey(parts[0]); err != nil {
return "", "", false
}
if err := ValidateKey(parts[1]); err != nil {
return "", "", false
}
return parts[0], parts[1], true
}
func (s *Server) s3Put(w http.ResponseWriter, r *http.Request, key string) {
release, ok := s.acceptWrite(w, r, true)
if !ok {
return
}
defer release()
contentType := r.Header.Get("Content-Type")
body, verifyPayload, ok := s.s3StreamingBody(w, r)
if !ok {
return
}
store, ok := s.store.(*FileStore)
if !ok {
s3Error(w, http.StatusInternalServerError, "InternalError", "store failed")
return
}
info, err := store.PutVerified(key, http.MaxBytesReader(w, body, s.cfg.MaxObjectSize), contentType, func() error {
if !verifyPayload() {
return errPayloadHashMismatch
}
return nil
})
if err != nil {
if maxBytesError(err) {
s3Error(w, http.StatusRequestEntityTooLarge, "EntityTooLarge", "object too large")
return
}
if errors.Is(err, errPayloadHashMismatch) {
s3Error(w, http.StatusForbidden, "XAmzContentSHA256Mismatch", "payload hash mismatch")
return
}
s3Error(w, http.StatusInternalServerError, "InternalError", "store failed")
return
}
w.Header().Set("ETag", quoteETag(info.ETag))
w.WriteHeader(http.StatusOK)
if r.Header.Get("X-Magpie-Replication") != "1" {
s.enqueueReplicationPut(key)
}
}
func (s *Server) s3Get(w http.ResponseWriter, r *http.Request, key string, publicRead bool) {
obj, err := s.store.Open(key)
if err != nil {
if errors.Is(err, ErrNotFound) {
s3Error(w, http.StatusNotFound, "NoSuchKey", "object not found")
return
}
s3Error(w, http.StatusInternalServerError, "InternalError", "read failed")
return
}
defer obj.Close()
s3ObjectHeaders(w, obj.Info)
if publicRead {
hardenPublicRead(w, obj.Info.ContentType)
} else if disposition := r.URL.Query().Get("response-content-disposition"); disposition != "" {
w.Header().Set("Content-Disposition", disposition)
}
http.ServeContent(w, r, obj.Info.Key, obj.Info.ModTime, obj)
}
func (s *Server) s3Head(w http.ResponseWriter, _ *http.Request, key string, publicRead bool) {
info, err := s.store.Stat(key)
if err != nil {
if errors.Is(err, ErrNotFound) {
w.WriteHeader(http.StatusNotFound)
return
}
w.WriteHeader(http.StatusInternalServerError)
return
}
s3ObjectHeaders(w, info)
if publicRead {
hardenPublicRead(w, info.ContentType)
}
w.WriteHeader(http.StatusOK)
}
// hardenPublicRead applies defense-in-depth headers to unauthenticated public
// reads. Objects carry an uploader-supplied Content-Type, so a public prefix
// could otherwise serve active content (HTML, SVG) that executes script in
// Magpie's own origin. The CSP sandbox disables scripting even when a browser
// navigates directly to such an object, and any content type that is not inert
// inline media is forced to download rather than render. The attacker-influenced
// response-content-disposition override is intentionally not honored here.
func hardenPublicRead(w http.ResponseWriter, contentType string) {
w.Header().Set("Content-Security-Policy", "default-src 'none'; sandbox")
if !inlineSafeContentType(contentType) {
w.Header().Set("Content-Disposition", "attachment")
}
}
// inlineSafeContentType reports whether a content type is inert raster/media
// that browsers render without executing script. SVG is excluded because it can
// carry script when loaded as a top-level document.
func inlineSafeContentType(contentType string) bool {
mediaType := contentType
if i := strings.IndexByte(mediaType, ';'); i >= 0 {
mediaType = mediaType[:i]
}
mediaType = strings.ToLower(strings.TrimSpace(mediaType))
if mediaType == "image/svg+xml" || mediaType == "image/svg" {
return false
}
return strings.HasPrefix(mediaType, "image/") ||
strings.HasPrefix(mediaType, "audio/") ||
strings.HasPrefix(mediaType, "video/")
}
func (s *Server) s3Delete(w http.ResponseWriter, r *http.Request, key string) {
if err := s.store.Delete(key); err != nil && !errors.Is(err, ErrNotFound) {
s3Error(w, http.StatusInternalServerError, "InternalError", "delete failed")
return
}
w.WriteHeader(http.StatusNoContent)
if r.Header.Get("X-Magpie-Replication") != "1" {
s.enqueueReplicationDelete(key)
}
}
func s3ObjectHeaders(w http.ResponseWriter, info ObjectInfo) {
contentType := info.ContentType
if contentType == "" {
contentType = "application/octet-stream"
}
w.Header().Set("Content-Type", contentType)
w.Header().Set("Content-Length", fmt.Sprintf("%d", info.Size))
w.Header().Set("ETag", quoteETag(info.ETag))
w.Header().Set("Last-Modified", info.ModTime.UTC().Format(http.TimeFormat))
w.Header().Set("Accept-Ranges", "bytes")
w.Header().Set("X-Content-Type-Options", "nosniff")
}
func quoteETag(etag string) string {
if etag == "" {
return "\"\""
}
if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") {
return etag
}
return "\"" + etag + "\""
}
type s3ErrorBody struct {
XMLName xml.Name `xml:"Error"`
Code string `xml:"Code"`
Message string `xml:"Message"`
}
func s3Error(w http.ResponseWriter, status int, code string, message string) {
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(status)
_ = xml.NewEncoder(w).Encode(s3ErrorBody{Code: code, Message: message})
}
func parseMaxKeys(value string) int {
if value == "" {
return 1000
}
parsed, err := strconv.Atoi(value)
if err != nil || parsed <= 0 {
return 1000
}
if parsed > 1000 {
return 1000
}
return parsed
}
func randomID() (string, error) {
buf := make([]byte, 18)
if _, err := rand.Read(buf); err != nil {
return "", err
}
return base64.RawURLEncoding.EncodeToString(buf), nil
}
func md5Hash() hashHash {
return md5.New()
}
type hashHash interface {
io.Writer
Sum([]byte) []byte
}
func (s *Server) multipartDir(uploadID string) string {
if !validUploadID(uploadID) {
return filepath.Join(s.store.(*FileStore).root, ".multipart", "invalid")
}
return filepath.Join(s.store.(*FileStore).root, ".multipart", uploadID)
}
func validUploadID(uploadID string) bool {
if uploadID == "" || len(uploadID) > 128 {
return false
}
for _, ch := range uploadID {
if (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || ch == '-' || ch == '_' {
continue
}
return false
}
return true
}
func (s *Server) startMultipartCleanup(ctx context.Context) {
store, ok := s.store.(*FileStore)
if !ok || s.cfg.MultipartMaxAge <= 0 {
return
}
if err := store.CleanupMultipart(s.cfg.MultipartMaxAge); err != nil {
log.Printf("multipart cleanup error=%v", err)
}
go func() {
ticker := time.NewTicker(time.Hour)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := store.CleanupMultipart(s.cfg.MultipartMaxAge); err != nil {
log.Printf("multipart cleanup error=%v", err)
}
}
}
}()
}
func (s *Server) s3BoundedBody(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, bool) {
body, verifyPayload, ok := s.s3StreamingBody(w, r)
if !ok {
return nil, false
}
data, err := io.ReadAll(http.MaxBytesReader(w, body, limit))
if err != nil {
if errors.Is(err, errPayloadHashMismatch) {
s3Error(w, http.StatusForbidden, "XAmzContentSHA256Mismatch", "payload hash mismatch")
return nil, false
}
s3Error(w, http.StatusRequestEntityTooLarge, "EntityTooLarge", "request body too large")
return nil, false
}
if !verifyPayload() {
s3Error(w, http.StatusForbidden, "XAmzContentSHA256Mismatch", "payload hash mismatch")
return nil, false
}
return data, true
}
func (s *Server) s3StreamingBody(w http.ResponseWriter, r *http.Request) (io.ReadCloser, func() bool, bool) {
payloadHash := r.Header.Get("X-Amz-Content-Sha256")
if payloadHash == "" || payloadHash == "UNSIGNED-PAYLOAD" {
return r.Body, func() bool { return true }, true
}
if len(payloadHash) != sha256.Size*2 {
s3Error(w, http.StatusBadRequest, "InvalidDigest", "unsupported payload hash")
return nil, nil, false
}
if _, err := hex.DecodeString(payloadHash); err != nil {
s3Error(w, http.StatusBadRequest, "InvalidDigest", "invalid payload hash")
return nil, nil, false
}
hash := sha256.New()
verified := false
return hashVerifyingReadCloser{reader: r.Body, hash: hash, expected: strings.ToLower(payloadHash)}, func() bool {
if verified {
return true
}
verified = hex.EncodeToString(hash.Sum(nil)) == strings.ToLower(payloadHash)
return verified
}, true
}
func signedHeaderIncludes(r *http.Request, header string) bool {
signedHeaders := r.URL.Query().Get("X-Amz-SignedHeaders")
if signedHeaders == "" {
auth := r.Header.Get("Authorization")
if strings.HasPrefix(auth, awsAlgorithm+" ") {
signedHeaders = parseAuthParams(strings.TrimPrefix(auth, awsAlgorithm+" "))["SignedHeaders"]
}
}
for _, signed := range strings.Split(signedHeaders, ";") {
if strings.EqualFold(strings.TrimSpace(signed), header) {
return true
}
}
return false
}
type hashVerifyingReadCloser struct {
reader io.ReadCloser
hash hashHash
expected string
}
func (r hashVerifyingReadCloser) Read(p []byte) (int, error) {
n, err := r.reader.Read(p)
if n > 0 {
_, _ = r.hash.Write(p[:n])
}
if errors.Is(err, io.EOF) && hex.EncodeToString(r.hash.Sum(nil)) != r.expected {
return n, errPayloadHashMismatch
}
return n, err
}
func (r hashVerifyingReadCloser) Close() error {
return r.reader.Close()
}
func (s *Server) multipartMeta(uploadID string) (multipartMeta, error) {
data, err := os.ReadFile(filepath.Join(s.multipartDir(uploadID), "meta.xml"))
if err != nil {
return multipartMeta{}, err
}
var meta multipartMeta
if err := xml.Unmarshal(data, &meta); err != nil {
return multipartMeta{}, err
}
return meta, nil
}
func sortParts(parts []completedPart) {
sort.Slice(parts, func(i, j int) bool { return parts[i].PartNumber < parts[j].PartNumber })
}
type s3ListBucketResult struct {
XMLName xml.Name `xml:"ListBucketResult"`
Xmlns string `xml:"xmlns,attr,omitempty"`
Name string `xml:"Name"`
Prefix string `xml:"Prefix"`
ContinuationToken string `xml:"ContinuationToken,omitempty"`
NextContinuationToken string `xml:"NextContinuationToken,omitempty"`
KeyCount int `xml:"KeyCount"`
MaxKeys int `xml:"MaxKeys"`
IsTruncated bool `xml:"IsTruncated"`
Contents []s3ListContent `xml:"Contents"`
}
type s3ListContent struct {
Key string `xml:"Key"`
LastModified string `xml:"LastModified"`
ETag string `xml:"ETag"`
Size int64 `xml:"Size"`
StorageClass string `xml:"StorageClass"`
}
type multipartMeta struct {
Key string `xml:"Key"`
ContentType string `xml:"ContentType"`
}
type createMultipartUploadResult struct {
XMLName xml.Name `xml:"InitiateMultipartUploadResult"`
UploadID string `xml:"UploadId"`
}
type completeMultipartUpload struct {
Parts []completedPart `xml:"Part"`
}
type completedPart struct {
PartNumber int `xml:"PartNumber"`
ETag string `xml:"ETag"`
}
type completeMultipartUploadResult struct {
XMLName xml.Name `xml:"CompleteMultipartUploadResult"`
Key string `xml:"Key"`
ETag string `xml:"ETag"`
}
type copyObjectResult struct {
XMLName xml.Name `xml:"CopyObjectResult"`
LastModified string `xml:"LastModified"`
ETag string `xml:"ETag"`
}
type deleteObjectsRequest struct {
Objects []deleteObjectRequest `xml:"Object"`
}
type deleteObjectRequest struct {
Key string `xml:"Key"`
}
type deleteObjectsResult struct {
XMLName xml.Name `xml:"DeleteResult"`
Deleted []deletedObject `xml:"Deleted"`
Errors []deleteObjectError `xml:"Error,omitempty"`
}
type deletedObject struct {
Key string `xml:"Key"`
}
type deleteObjectError struct {
Key string `xml:"Key"`
Code string `xml:"Code"`
Message string `xml:"Message"`
}