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.
410 lines
15 KiB
410 lines
15 KiB
package s3api
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
|
)
|
|
|
|
// S3MultipartIAMManager handles IAM integration for multipart upload operations
|
|
type S3MultipartIAMManager struct {
|
|
s3iam *S3IAMIntegration
|
|
}
|
|
|
|
// NewS3MultipartIAMManager creates a new multipart IAM manager
|
|
func NewS3MultipartIAMManager(s3iam *S3IAMIntegration) *S3MultipartIAMManager {
|
|
return &S3MultipartIAMManager{
|
|
s3iam: s3iam,
|
|
}
|
|
}
|
|
|
|
// MultipartUploadRequest represents a multipart upload request
|
|
type MultipartUploadRequest struct {
|
|
Bucket string `json:"bucket"` // S3 bucket name
|
|
ObjectKey string `json:"object_key"` // S3 object key
|
|
UploadID string `json:"upload_id"` // Multipart upload ID
|
|
PartNumber int `json:"part_number"` // Part number for upload part
|
|
Operation string `json:"operation"` // Multipart operation type
|
|
SessionToken string `json:"session_token"` // JWT session token
|
|
Headers map[string]string `json:"headers"` // Request headers
|
|
ContentSize int64 `json:"content_size"` // Content size for validation
|
|
}
|
|
|
|
// MultipartUploadPolicy represents security policies for multipart uploads
|
|
type MultipartUploadPolicy struct {
|
|
MaxPartSize int64 `json:"max_part_size"` // Maximum part size (5GB AWS limit)
|
|
MinPartSize int64 `json:"min_part_size"` // Minimum part size (5MB AWS limit, except last part)
|
|
MaxParts int `json:"max_parts"` // Maximum number of parts (10,000 AWS limit)
|
|
MaxUploadDuration time.Duration `json:"max_upload_duration"` // Maximum time to complete multipart upload
|
|
AllowedContentTypes []string `json:"allowed_content_types"` // Allowed content types
|
|
RequiredHeaders []string `json:"required_headers"` // Required headers for validation
|
|
IPWhitelist []string `json:"ip_whitelist"` // Allowed IP addresses/ranges
|
|
}
|
|
|
|
// MultipartOperation represents different multipart upload operations
|
|
type MultipartOperation string
|
|
|
|
const (
|
|
MultipartOpInitiate MultipartOperation = "initiate"
|
|
MultipartOpUploadPart MultipartOperation = "upload_part"
|
|
MultipartOpComplete MultipartOperation = "complete"
|
|
MultipartOpAbort MultipartOperation = "abort"
|
|
MultipartOpList MultipartOperation = "list"
|
|
MultipartOpListParts MultipartOperation = "list_parts"
|
|
)
|
|
|
|
// ValidateMultipartOperationWithIAM validates multipart operations using IAM policies
|
|
func (iam *IdentityAccessManagement) ValidateMultipartOperationWithIAM(r *http.Request, identity *Identity, operation MultipartOperation) s3err.ErrorCode {
|
|
if iam.iamIntegration == nil {
|
|
// Fall back to standard validation
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
// Extract bucket and object from request
|
|
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
|
|
// Determine the S3 action based on multipart operation
|
|
action := determineMultipartS3Action(operation)
|
|
|
|
// Extract session token from request
|
|
sessionToken := extractSessionTokenFromRequest(r)
|
|
if sessionToken == "" {
|
|
// No session token - use standard auth
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
// Create IAM identity for authorization
|
|
principalArn := fmt.Sprintf("arn:seaweed:sts::assumed-role/MultipartUser/%s", identity.Name)
|
|
iamIdentity := &IAMIdentity{
|
|
Name: identity.Name,
|
|
Principal: principalArn,
|
|
SessionToken: sessionToken,
|
|
Account: identity.Account,
|
|
}
|
|
|
|
// Authorize using IAM
|
|
ctx := r.Context()
|
|
errCode := iam.iamIntegration.AuthorizeAction(ctx, iamIdentity, action, bucket, object, r)
|
|
if errCode != s3err.ErrNone {
|
|
glog.V(3).Infof("IAM authorization failed for multipart operation: principal=%s operation=%s action=%s bucket=%s object=%s",
|
|
iamIdentity.Principal, operation, action, bucket, object)
|
|
return errCode
|
|
}
|
|
|
|
glog.V(3).Infof("IAM authorization succeeded for multipart operation: principal=%s operation=%s action=%s bucket=%s object=%s",
|
|
iamIdentity.Principal, operation, action, bucket, object)
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
// ValidateMultipartRequestWithPolicy validates multipart request against security policy
|
|
func (policy *MultipartUploadPolicy) ValidateMultipartRequestWithPolicy(req *MultipartUploadRequest) error {
|
|
if req == nil {
|
|
return fmt.Errorf("multipart request cannot be nil")
|
|
}
|
|
|
|
// Validate part size for upload part operations
|
|
if req.Operation == string(MultipartOpUploadPart) {
|
|
if req.ContentSize > policy.MaxPartSize {
|
|
return fmt.Errorf("part size %d exceeds maximum allowed %d", req.ContentSize, policy.MaxPartSize)
|
|
}
|
|
|
|
// Minimum part size validation (except for last part)
|
|
// Note: Last part validation would require knowing if this is the final part
|
|
if req.ContentSize < policy.MinPartSize && req.ContentSize > 0 {
|
|
glog.V(2).Infof("Part size %d is below minimum %d - assuming last part", req.ContentSize, policy.MinPartSize)
|
|
}
|
|
|
|
// Validate part number
|
|
if req.PartNumber < 1 || req.PartNumber > policy.MaxParts {
|
|
return fmt.Errorf("part number %d is invalid (must be 1-%d)", req.PartNumber, policy.MaxParts)
|
|
}
|
|
}
|
|
|
|
// Validate required headers first
|
|
if req.Headers != nil {
|
|
for _, requiredHeader := range policy.RequiredHeaders {
|
|
if _, exists := req.Headers[requiredHeader]; !exists {
|
|
// Check lowercase version
|
|
if _, exists := req.Headers[strings.ToLower(requiredHeader)]; !exists {
|
|
return fmt.Errorf("required header %s is missing", requiredHeader)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Validate content type if specified
|
|
if len(policy.AllowedContentTypes) > 0 && req.Headers != nil {
|
|
contentType := req.Headers["Content-Type"]
|
|
if contentType == "" {
|
|
contentType = req.Headers["content-type"]
|
|
}
|
|
|
|
allowed := false
|
|
for _, allowedType := range policy.AllowedContentTypes {
|
|
if contentType == allowedType {
|
|
allowed = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !allowed {
|
|
return fmt.Errorf("content type %s is not allowed", contentType)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Enhanced multipart handlers with IAM integration
|
|
|
|
// NewMultipartUploadWithIAM handles initiate multipart upload with IAM validation
|
|
func (s3a *S3ApiServer) NewMultipartUploadWithIAM(w http.ResponseWriter, r *http.Request) {
|
|
// Validate IAM permissions first
|
|
if s3a.iam.iamIntegration != nil {
|
|
if identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_WRITE); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
} else {
|
|
// Additional multipart-specific IAM validation
|
|
if errCode := s3a.iam.ValidateMultipartOperationWithIAM(r, identity, MultipartOpInitiate); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delegate to existing handler
|
|
s3a.NewMultipartUploadHandler(w, r)
|
|
}
|
|
|
|
// CompleteMultipartUploadWithIAM handles complete multipart upload with IAM validation
|
|
func (s3a *S3ApiServer) CompleteMultipartUploadWithIAM(w http.ResponseWriter, r *http.Request) {
|
|
// Validate IAM permissions first
|
|
if s3a.iam.iamIntegration != nil {
|
|
if identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_WRITE); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
} else {
|
|
// Additional multipart-specific IAM validation
|
|
if errCode := s3a.iam.ValidateMultipartOperationWithIAM(r, identity, MultipartOpComplete); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delegate to existing handler
|
|
s3a.CompleteMultipartUploadHandler(w, r)
|
|
}
|
|
|
|
// AbortMultipartUploadWithIAM handles abort multipart upload with IAM validation
|
|
func (s3a *S3ApiServer) AbortMultipartUploadWithIAM(w http.ResponseWriter, r *http.Request) {
|
|
// Validate IAM permissions first
|
|
if s3a.iam.iamIntegration != nil {
|
|
if identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_WRITE); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
} else {
|
|
// Additional multipart-specific IAM validation
|
|
if errCode := s3a.iam.ValidateMultipartOperationWithIAM(r, identity, MultipartOpAbort); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delegate to existing handler
|
|
s3a.AbortMultipartUploadHandler(w, r)
|
|
}
|
|
|
|
// ListMultipartUploadsWithIAM handles list multipart uploads with IAM validation
|
|
func (s3a *S3ApiServer) ListMultipartUploadsWithIAM(w http.ResponseWriter, r *http.Request) {
|
|
// Validate IAM permissions first
|
|
if s3a.iam.iamIntegration != nil {
|
|
if identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_LIST); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
} else {
|
|
// Additional multipart-specific IAM validation
|
|
if errCode := s3a.iam.ValidateMultipartOperationWithIAM(r, identity, MultipartOpList); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delegate to existing handler
|
|
s3a.ListMultipartUploadsHandler(w, r)
|
|
}
|
|
|
|
// UploadPartWithIAM handles upload part with IAM validation
|
|
func (s3a *S3ApiServer) UploadPartWithIAM(w http.ResponseWriter, r *http.Request) {
|
|
// Validate IAM permissions first
|
|
if s3a.iam.iamIntegration != nil {
|
|
if identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_WRITE); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
} else {
|
|
// Additional multipart-specific IAM validation
|
|
if errCode := s3a.iam.ValidateMultipartOperationWithIAM(r, identity, MultipartOpUploadPart); errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
|
|
// Validate part size and other policies
|
|
if err := s3a.validateUploadPartRequest(r); err != nil {
|
|
glog.Errorf("Upload part validation failed: %v", err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delegate to existing object PUT handler (which handles upload part)
|
|
s3a.PutObjectHandler(w, r)
|
|
}
|
|
|
|
// Helper functions
|
|
|
|
// determineMultipartS3Action maps multipart operations to S3 actions
|
|
func determineMultipartS3Action(operation MultipartOperation) Action {
|
|
switch operation {
|
|
case MultipartOpInitiate:
|
|
return s3_constants.ACTION_WRITE // s3:CreateMultipartUpload maps to WRITE
|
|
case MultipartOpUploadPart:
|
|
return s3_constants.ACTION_WRITE // s3:UploadPart maps to WRITE
|
|
case MultipartOpComplete:
|
|
return s3_constants.ACTION_WRITE // s3:CompleteMultipartUpload maps to WRITE
|
|
case MultipartOpAbort:
|
|
return s3_constants.ACTION_WRITE // s3:AbortMultipartUpload maps to WRITE
|
|
case MultipartOpList:
|
|
return s3_constants.ACTION_LIST // s3:ListMultipartUploads maps to LIST
|
|
case MultipartOpListParts:
|
|
return s3_constants.ACTION_LIST // s3:ListParts maps to LIST
|
|
default:
|
|
return s3_constants.ACTION_READ // Default fallback
|
|
}
|
|
}
|
|
|
|
// extractSessionTokenFromRequest extracts session token from various request sources
|
|
func extractSessionTokenFromRequest(r *http.Request) string {
|
|
// Check Authorization header for Bearer token
|
|
if authHeader := r.Header.Get("Authorization"); authHeader != "" {
|
|
if strings.HasPrefix(authHeader, "Bearer ") {
|
|
return strings.TrimPrefix(authHeader, "Bearer ")
|
|
}
|
|
}
|
|
|
|
// Check X-Amz-Security-Token header
|
|
if token := r.Header.Get("X-Amz-Security-Token"); token != "" {
|
|
return token
|
|
}
|
|
|
|
// Check query parameters for presigned URL tokens
|
|
if token := r.URL.Query().Get("X-Amz-Security-Token"); token != "" {
|
|
return token
|
|
}
|
|
|
|
return ""
|
|
}
|
|
|
|
// validateUploadPartRequest validates upload part request against policies
|
|
func (s3a *S3ApiServer) validateUploadPartRequest(r *http.Request) error {
|
|
// Get default multipart policy
|
|
policy := DefaultMultipartUploadPolicy()
|
|
|
|
// Extract part number from query
|
|
partNumberStr := r.URL.Query().Get("partNumber")
|
|
if partNumberStr == "" {
|
|
return fmt.Errorf("missing partNumber parameter")
|
|
}
|
|
|
|
partNumber, err := strconv.Atoi(partNumberStr)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid partNumber: %v", err)
|
|
}
|
|
|
|
// Get content length
|
|
contentLength := r.ContentLength
|
|
if contentLength < 0 {
|
|
contentLength = 0
|
|
}
|
|
|
|
// Create multipart request for validation
|
|
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
multipartReq := &MultipartUploadRequest{
|
|
Bucket: bucket,
|
|
ObjectKey: object,
|
|
PartNumber: partNumber,
|
|
Operation: string(MultipartOpUploadPart),
|
|
ContentSize: contentLength,
|
|
Headers: make(map[string]string),
|
|
}
|
|
|
|
// Copy relevant headers
|
|
for key, values := range r.Header {
|
|
if len(values) > 0 {
|
|
multipartReq.Headers[key] = values[0]
|
|
}
|
|
}
|
|
|
|
// Validate against policy
|
|
return policy.ValidateMultipartRequestWithPolicy(multipartReq)
|
|
}
|
|
|
|
// DefaultMultipartUploadPolicy returns a default multipart upload security policy
|
|
func DefaultMultipartUploadPolicy() *MultipartUploadPolicy {
|
|
return &MultipartUploadPolicy{
|
|
MaxPartSize: 5 * 1024 * 1024 * 1024, // 5GB AWS limit
|
|
MinPartSize: 5 * 1024 * 1024, // 5MB AWS minimum (except last part)
|
|
MaxParts: 10000, // AWS limit
|
|
MaxUploadDuration: 7 * 24 * time.Hour, // 7 days to complete upload
|
|
AllowedContentTypes: []string{}, // Empty means all types allowed
|
|
RequiredHeaders: []string{}, // No required headers by default
|
|
IPWhitelist: []string{}, // Empty means no IP restrictions
|
|
}
|
|
}
|
|
|
|
// MultipartUploadSession represents an ongoing multipart upload session
|
|
type MultipartUploadSession struct {
|
|
UploadID string `json:"upload_id"`
|
|
Bucket string `json:"bucket"`
|
|
ObjectKey string `json:"object_key"`
|
|
Initiator string `json:"initiator"` // User who initiated the upload
|
|
Owner string `json:"owner"` // Object owner
|
|
CreatedAt time.Time `json:"created_at"` // When upload was initiated
|
|
Parts []MultipartUploadPart `json:"parts"` // Uploaded parts
|
|
Metadata map[string]string `json:"metadata"` // Object metadata
|
|
Policy *MultipartUploadPolicy `json:"policy"` // Applied security policy
|
|
SessionToken string `json:"session_token"` // IAM session token
|
|
}
|
|
|
|
// MultipartUploadPart represents an uploaded part
|
|
type MultipartUploadPart struct {
|
|
PartNumber int `json:"part_number"`
|
|
Size int64 `json:"size"`
|
|
ETag string `json:"etag"`
|
|
LastModified time.Time `json:"last_modified"`
|
|
Checksum string `json:"checksum"` // Optional integrity checksum
|
|
}
|
|
|
|
// GetMultipartUploadSessions retrieves active multipart upload sessions for a bucket
|
|
func (s3a *S3ApiServer) GetMultipartUploadSessions(bucket string) ([]*MultipartUploadSession, error) {
|
|
// This would typically query the filer for active multipart uploads
|
|
// For now, return empty list as this is a placeholder for the full implementation
|
|
return []*MultipartUploadSession{}, nil
|
|
}
|
|
|
|
// CleanupExpiredMultipartUploads removes expired multipart upload sessions
|
|
func (s3a *S3ApiServer) CleanupExpiredMultipartUploads(maxAge time.Duration) error {
|
|
// This would typically scan for and remove expired multipart uploads
|
|
// Implementation would depend on how multipart sessions are stored in the filer
|
|
glog.V(2).Infof("Cleanup expired multipart uploads older than %v", maxAge)
|
|
return nil
|
|
}
|