/* * Minio Cloud Storage, (C) 2017 Minio, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package cmd import ( "encoding/base64" "encoding/hex" "fmt" "hash" "io" "net/http" "net/url" "strconv" "strings" "sync" "time" "github.com/Azure/azure-sdk-for-go/storage" "github.com/minio/minio-go/pkg/policy" "github.com/minio/sha256-simd" ) const globalAzureAPIVersion = "2016-05-31" // To store metadata during NewMultipartUpload which will be used after // CompleteMultipartUpload to call SetBlobMetadata. type azureMultipartMetaInfo struct { meta map[string]map[string]string *sync.Mutex } // Return metadata map of the multipart object. func (a *azureMultipartMetaInfo) get(key string) map[string]string { a.Lock() defer a.Unlock() return a.meta[key] } // Set metadata map for the multipart object. func (a *azureMultipartMetaInfo) set(key string, value map[string]string) { a.Lock() defer a.Unlock() a.meta[key] = value } // Delete metadata map for the multipart object. func (a *azureMultipartMetaInfo) del(key string) { a.Lock() defer a.Unlock() delete(a.meta, key) } // azureObjects - Implements Object layer for Azure blob storage. type azureObjects struct { client storage.BlobStorageClient // Azure sdk client metaInfo azureMultipartMetaInfo } // Convert azure errors to minio object layer errors. func azureToObjectError(err error, params ...string) error { if err == nil { return nil } e, ok := err.(*Error) if !ok { // Code should be fixed if this function is called without doing traceError() // Else handling different situations in this function makes this function complicated. errorIf(err, "Expected type *Error") return err } err = e.e bucket := "" object := "" if len(params) >= 1 { bucket = params[0] } if len(params) == 2 { object = params[1] } azureErr, ok := err.(storage.AzureStorageServiceError) if !ok { // We don't interpret non Azure errors. As azure errors will // have StatusCode to help to convert to object errors. return e } switch azureErr.Code { case "ContainerAlreadyExists": err = BucketExists{Bucket: bucket} case "InvalidResourceName": err = BucketNameInvalid{Bucket: bucket} default: switch azureErr.StatusCode { case http.StatusNotFound: if object != "" { err = ObjectNotFound{bucket, object} } else { err = BucketNotFound{Bucket: bucket} } case http.StatusBadRequest: err = BucketNameInvalid{Bucket: bucket} } } e.e = err return e } // Inits azure blob storage client and returns azureObjects. func newAzureLayer(endPoint string, account, key string, secure bool) (GatewayLayer, error) { if endPoint == "" { endPoint = storage.DefaultBaseURL } c, err := storage.NewClient(account, key, endPoint, globalAzureAPIVersion, secure) if err != nil { return &azureObjects{}, err } return &azureObjects{ client: c.GetBlobService(), metaInfo: azureMultipartMetaInfo{ meta: make(map[string]map[string]string), Mutex: &sync.Mutex{}, }, }, nil } // Shutdown - save any gateway metadata to disk // if necessary and reload upon next restart. func (a *azureObjects) Shutdown() error { // TODO return nil } // StorageInfo - Not relevant to Azure backend. func (a *azureObjects) StorageInfo() StorageInfo { return StorageInfo{} } // MakeBucket - Create a new container on azure backend. func (a *azureObjects) MakeBucket(bucket string) error { // will never be called, only satisfy ObjectLayer interface return traceError(NotImplemented{}) } // MakeBucketWithLocation - Create a new container on azure backend. func (a *azureObjects) MakeBucketWithLocation(bucket, location string) error { err := a.client.CreateContainer(bucket, storage.ContainerAccessTypePrivate) return azureToObjectError(traceError(err), bucket) } // GetBucketInfo - Get bucket metadata.. func (a *azureObjects) GetBucketInfo(bucket string) (BucketInfo, error) { // Azure does not have an equivalent call, hence use ListContainers. resp, err := a.client.ListContainers(storage.ListContainersParameters{ Prefix: bucket, }) if err != nil { return BucketInfo{}, azureToObjectError(traceError(err), bucket) } for _, container := range resp.Containers { if container.Name == bucket { t, e := time.Parse(time.RFC1123, container.Properties.LastModified) if e == nil { return BucketInfo{ Name: bucket, Created: t, }, nil } // else continue } } return BucketInfo{}, traceError(BucketNotFound{Bucket: bucket}) } // ListBuckets - Lists all azure containers, uses Azure equivalent ListContainers. func (a *azureObjects) ListBuckets() (buckets []BucketInfo, err error) { resp, err := a.client.ListContainers(storage.ListContainersParameters{}) if err != nil { return nil, azureToObjectError(traceError(err)) } for _, container := range resp.Containers { t, e := time.Parse(time.RFC1123, container.Properties.LastModified) if e != nil { return nil, traceError(e) } buckets = append(buckets, BucketInfo{ Name: container.Name, Created: t, }) } return buckets, nil } // DeleteBucket - delete a container on azure, uses Azure equivalent DeleteContainer. func (a *azureObjects) DeleteBucket(bucket string) error { return azureToObjectError(traceError(a.client.DeleteContainer(bucket)), bucket) } // ListObjects - lists all blobs on azure with in a container filtered by prefix // and marker, uses Azure equivalent ListBlobs. func (a *azureObjects) ListObjects(bucket, prefix, marker, delimiter string, maxKeys int) (result ListObjectsInfo, err error) { resp, err := a.client.ListBlobs(bucket, storage.ListBlobsParameters{ Prefix: prefix, Marker: marker, Delimiter: delimiter, MaxResults: uint(maxKeys), }) if err != nil { return result, azureToObjectError(traceError(err), bucket, prefix) } result.IsTruncated = resp.NextMarker != "" result.NextMarker = resp.NextMarker for _, object := range resp.Blobs { t, e := time.Parse(time.RFC1123, object.Properties.LastModified) if e != nil { continue } result.Objects = append(result.Objects, ObjectInfo{ Bucket: bucket, Name: object.Name, ModTime: t, Size: object.Properties.ContentLength, ETag: canonicalizeETag(object.Properties.Etag), ContentType: object.Properties.ContentType, ContentEncoding: object.Properties.ContentEncoding, }) } result.Prefixes = resp.BlobPrefixes return result, nil } // GetObject - reads an object from azure. Supports additional // parameters like offset and length which are synonymous with // HTTP Range requests. // // startOffset indicates the starting read location of the object. // length indicates the total length of the object. func (a *azureObjects) GetObject(bucket, object string, startOffset int64, length int64, writer io.Writer) error { byteRange := fmt.Sprintf("%d-", startOffset) if length > 0 && startOffset > 0 { byteRange = fmt.Sprintf("%d-%d", startOffset, startOffset+length-1) } var rc io.ReadCloser var err error if startOffset == 0 && length == 0 { rc, err = a.client.GetBlob(bucket, object) } else { rc, err = a.client.GetBlobRange(bucket, object, byteRange, nil) } if err != nil { return azureToObjectError(traceError(err), bucket, object) } _, err = io.Copy(writer, rc) rc.Close() return traceError(err) } // GetObjectInfo - reads blob metadata properties and replies back ObjectInfo, // uses zure equivalent GetBlobProperties. func (a *azureObjects) GetObjectInfo(bucket, object string) (objInfo ObjectInfo, err error) { prop, err := a.client.GetBlobProperties(bucket, object) if err != nil { return objInfo, azureToObjectError(traceError(err), bucket, object) } t, err := time.Parse(time.RFC1123, prop.LastModified) if err != nil { return objInfo, traceError(err) } objInfo = ObjectInfo{ Bucket: bucket, UserDefined: make(map[string]string), ETag: canonicalizeETag(prop.Etag), ModTime: t, Name: object, Size: prop.ContentLength, } if prop.ContentEncoding != "" { objInfo.UserDefined["Content-Encoding"] = prop.ContentEncoding } objInfo.UserDefined["Content-Type"] = prop.ContentType return objInfo, nil } // Canonicalize the metadata headers, without this azure-sdk calculates // incorrect signature. This attempt to canonicalize is to convert // any HTTP header which is of form say `accept-encoding` should be // converted to `Accept-Encoding` in its canonical form. func canonicalMetadata(metadata map[string]string) (canonical map[string]string) { canonical = make(map[string]string) for k, v := range metadata { canonical[http.CanonicalHeaderKey(k)] = v } return canonical } // PutObject - Create a new blob with the incoming data, // uses Azure equivalent CreateBlockBlobFromReader. func (a *azureObjects) PutObject(bucket, object string, size int64, data io.Reader, metadata map[string]string, sha256sum string) (objInfo ObjectInfo, err error) { var sha256Writer hash.Hash teeReader := data if sha256sum != "" { sha256Writer = sha256.New() teeReader = io.TeeReader(data, sha256Writer) } delete(metadata, "etag") err = a.client.CreateBlockBlobFromReader(bucket, object, uint64(size), teeReader, canonicalMetadata(metadata)) if err != nil { return objInfo, azureToObjectError(traceError(err), bucket, object) } if sha256sum != "" { newSHA256sum := hex.EncodeToString(sha256Writer.Sum(nil)) if newSHA256sum != sha256sum { a.client.DeleteBlob(bucket, object, nil) return ObjectInfo{}, traceError(SHA256Mismatch{}) } } return a.GetObjectInfo(bucket, object) } // CopyObject - Copies a blob from source container to destination container. // Uses Azure equivalent CopyBlob API. func (a *azureObjects) CopyObject(srcBucket, srcObject, destBucket, destObject string, metadata map[string]string) (objInfo ObjectInfo, err error) { err = a.client.CopyBlob(destBucket, destObject, a.client.GetBlobURL(srcBucket, srcObject)) if err != nil { return objInfo, azureToObjectError(traceError(err), srcBucket, srcObject) } return a.GetObjectInfo(destBucket, destObject) } // DeleteObject - Deletes a blob on azure container, uses Azure // equivalent DeleteBlob API. func (a *azureObjects) DeleteObject(bucket, object string) error { err := a.client.DeleteBlob(bucket, object, nil) if err != nil { return azureToObjectError(traceError(err), bucket, object) } return nil } // ListMultipartUploads - Incomplete implementation, for now just return the prefix if it is an incomplete upload. // FIXME: Full ListMultipartUploads is not supported yet. It is supported just enough to help our client libs to // support re-uploads. a.client.ListBlobs() can be made to return entries which include uncommitted blobs using // which we need to filter out the committed blobs to get the list of uncommitted blobs. func (a *azureObjects) ListMultipartUploads(bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) { result.MaxUploads = maxUploads result.Prefix = prefix result.Delimiter = delimiter meta := a.metaInfo.get(prefix) if meta == nil { // In case minio was restarted after NewMultipartUpload and before CompleteMultipartUpload we expect // the client to do a fresh upload so that any metadata like content-type are sent again in the // NewMultipartUpload. return result, nil } result.Uploads = []uploadMetadata{{prefix, prefix, UTCNow(), "", nil}} return result, nil } // NewMultipartUpload - Use Azure equivalent CreateBlockBlob. func (a *azureObjects) NewMultipartUpload(bucket, object string, metadata map[string]string) (uploadID string, err error) { // Azure doesn't return a unique upload ID and we use object name in place of it. Azure allows multiple uploads to // co-exist as long as the user keeps the blocks uploaded (in block blobs) unique amongst concurrent upload attempts. // Each concurrent client, keeps its own blockID list which it can commit. uploadID = object if metadata == nil { // Store an empty map as a placeholder else ListObjectParts/PutObjectPart will not work properly. metadata = make(map[string]string) } else { metadata = canonicalMetadata(metadata) } a.metaInfo.set(uploadID, metadata) return uploadID, nil } // CopyObjectPart - Not implemented. func (a *azureObjects) CopyObjectPart(srcBucket, srcObject, destBucket, destObject string, uploadID string, partID int, startOffset int64, length int64) (info PartInfo, err error) { return info, traceError(NotImplemented{}) } // Encode partID+md5Hex to a blockID. func azureGetBlockID(partID int, md5Hex string) string { return base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%.5d.%s", partID, md5Hex))) } // Decode blockID to partID+md5Hex. func azureParseBlockID(blockID string) (int, string, error) { idByte, err := base64.StdEncoding.DecodeString(blockID) if err != nil { return 0, "", traceError(err) } idStr := string(idByte) splitRes := strings.Split(idStr, ".") if len(splitRes) != 2 { return 0, "", traceError(errUnexpected) } partID, err := strconv.Atoi(splitRes[0]) if err != nil { return 0, "", traceError(err) } return partID, splitRes[1], nil } // PutObjectPart - Use Azure equivalent PutBlockWithLength. func (a *azureObjects) PutObjectPart(bucket, object, uploadID string, partID int, size int64, data io.Reader, md5Hex string, sha256sum string) (info PartInfo, err error) { if meta := a.metaInfo.get(uploadID); meta == nil { return info, traceError(InvalidUploadID{}) } var sha256Writer hash.Hash if sha256sum != "" { sha256Writer = sha256.New() } teeReader := io.TeeReader(data, sha256Writer) id := azureGetBlockID(partID, md5Hex) err = a.client.PutBlockWithLength(bucket, object, id, uint64(size), teeReader, nil) if err != nil { return info, azureToObjectError(traceError(err), bucket, object) } if sha256sum != "" { newSHA256sum := hex.EncodeToString(sha256Writer.Sum(nil)) if newSHA256sum != sha256sum { return PartInfo{}, traceError(SHA256Mismatch{}) } } info.PartNumber = partID info.ETag = md5Hex info.LastModified = UTCNow() info.Size = size return info, nil } // ListObjectParts - Use Azure equivalent GetBlockList. func (a *azureObjects) ListObjectParts(bucket, object, uploadID string, partNumberMarker int, maxParts int) (result ListPartsInfo, err error) { result.Bucket = bucket result.Object = object result.UploadID = uploadID result.MaxParts = maxParts if meta := a.metaInfo.get(uploadID); meta == nil { return result, nil } resp, err := a.client.GetBlockList(bucket, object, storage.BlockListTypeUncommitted) if err != nil { return result, azureToObjectError(traceError(err), bucket, object) } tmpMaxParts := 0 partCount := 0 // Used for figuring out IsTruncated. nextPartNumberMarker := 0 for _, part := range resp.UncommittedBlocks { if tmpMaxParts == maxParts { // Also takes care of the case if maxParts = 0 break } partCount++ partID, md5Hex, err := azureParseBlockID(part.Name) if err != nil { return result, err } if partID <= partNumberMarker { continue } result.Parts = append(result.Parts, PartInfo{ partID, UTCNow(), md5Hex, part.Size, }) tmpMaxParts++ nextPartNumberMarker = partID } if partCount < len(resp.UncommittedBlocks) { result.IsTruncated = true result.NextPartNumberMarker = nextPartNumberMarker } return result, nil } // AbortMultipartUpload - Not Implemented. // There is no corresponding API in azure to abort an incomplete upload. The uncommmitted blocks // gets deleted after one week. func (a *azureObjects) AbortMultipartUpload(bucket, object, uploadID string) error { a.metaInfo.del(uploadID) return nil } // CompleteMultipartUpload - Use Azure equivalent PutBlockList. func (a *azureObjects) CompleteMultipartUpload(bucket, object, uploadID string, uploadedParts []completePart) (objInfo ObjectInfo, err error) { meta := a.metaInfo.get(uploadID) if meta == nil { return objInfo, traceError(InvalidUploadID{uploadID}) } var blocks []storage.Block for _, part := range uploadedParts { blocks = append(blocks, storage.Block{ ID: azureGetBlockID(part.PartNumber, part.ETag), Status: storage.BlockStatusUncommitted, }) } err = a.client.PutBlockList(bucket, object, blocks) if err != nil { return objInfo, azureToObjectError(traceError(err), bucket, object) } if len(meta) > 0 { prop := storage.BlobHeaders{ ContentMD5: meta["Content-Md5"], ContentLanguage: meta["Content-Language"], ContentEncoding: meta["Content-Encoding"], ContentType: meta["Content-Type"], CacheControl: meta["Cache-Control"], } err = a.client.SetBlobProperties(bucket, object, prop) if err != nil { return objInfo, azureToObjectError(traceError(err), bucket, object) } } a.metaInfo.del(uploadID) return a.GetObjectInfo(bucket, object) } func anonErrToObjectErr(statusCode int, params ...string) error { bucket := "" object := "" if len(params) >= 1 { bucket = params[0] } if len(params) == 2 { object = params[1] } switch statusCode { case http.StatusNotFound: if object != "" { return ObjectNotFound{bucket, object} } return BucketNotFound{Bucket: bucket} case http.StatusBadRequest: if object != "" { return ObjectNameInvalid{bucket, object} } return BucketNameInvalid{Bucket: bucket} } return errUnexpected } // Copied from github.com/Azure/azure-sdk-for-go/storage/blob.go func azureListBlobsGetParameters(p storage.ListBlobsParameters) url.Values { out := url.Values{} if p.Prefix != "" { out.Set("prefix", p.Prefix) } if p.Delimiter != "" { out.Set("delimiter", p.Delimiter) } if p.Marker != "" { out.Set("marker", p.Marker) } if p.Include != "" { out.Set("include", p.Include) } if p.MaxResults != 0 { out.Set("maxresults", fmt.Sprintf("%v", p.MaxResults)) } if p.Timeout != 0 { out.Set("timeout", fmt.Sprintf("%v", p.Timeout)) } return out } // SetBucketPolicies - Azure supports three types of container policies: // storage.ContainerAccessTypeContainer - readonly in minio terminology // storage.ContainerAccessTypeBlob - readonly without listing in minio terminology // storage.ContainerAccessTypePrivate - none in minio terminology // As the common denominator for minio and azure is readonly and none, we support // these two policies at the bucket level. func (a *azureObjects) SetBucketPolicies(bucket string, policyInfo policy.BucketAccessPolicy) error { var policies []BucketAccessPolicy for prefix, policy := range policy.GetPolicies(policyInfo.Statements, bucket) { policies = append(policies, BucketAccessPolicy{ Prefix: prefix, Policy: policy, }) } prefix := bucket + "/*" // For all objects inside the bucket. if len(policies) != 1 { return traceError(NotImplemented{}) } if policies[0].Prefix != prefix { return traceError(NotImplemented{}) } if policies[0].Policy != policy.BucketPolicyReadOnly { return traceError(NotImplemented{}) } perm := storage.ContainerPermissions{ AccessType: storage.ContainerAccessTypeContainer, AccessPolicies: nil, } err := a.client.SetContainerPermissions(bucket, perm, 0, "") return azureToObjectError(traceError(err), bucket) } // GetBucketPolicies - Get the container ACL and convert it to canonical []bucketAccessPolicy func (a *azureObjects) GetBucketPolicies(bucket string) (policy.BucketAccessPolicy, error) { policyInfo := policy.BucketAccessPolicy{Version: "2012-10-17"} perm, err := a.client.GetContainerPermissions(bucket, 0, "") if err != nil { return policy.BucketAccessPolicy{}, azureToObjectError(traceError(err), bucket) } switch perm.AccessType { case storage.ContainerAccessTypePrivate: // Do nothing case storage.ContainerAccessTypeContainer: policyInfo.Statements = policy.SetPolicy(policyInfo.Statements, policy.BucketPolicyReadOnly, bucket, "") default: return policy.BucketAccessPolicy{}, azureToObjectError(traceError(NotImplemented{})) } return policyInfo, nil } // DeleteBucketPolicies - Set the container ACL to "private" func (a *azureObjects) DeleteBucketPolicies(bucket string) error { perm := storage.ContainerPermissions{ AccessType: storage.ContainerAccessTypePrivate, AccessPolicies: nil, } err := a.client.SetContainerPermissions(bucket, perm, 0, "") return azureToObjectError(traceError(err)) }