oss.go

//
// oss-aliyun-go - Go packages to interact with the Open Storage Service.
//
//   https://wiki.ubuntu.com/goamz
//
// Copyright (c) 2011 Canonical Ltd.
//

package oss

import (
	"bytes"
	"encoding/xml"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"strconv"
	"strings"
	"time"
)

const (
	debug       = false
	DefaultHost = "http://oss.aliyuncs.com"

	// https://docs.aliyun.com/#/pub/oss/product-documentation/domain-region
	// https://help.aliyun.com/knowledge_detail/5989609.html
	HangZhou            = "oss-cn-hangzhou"
	QingDao             = "oss-cn-qingdao"
	BeiJing             = "oss-cn-beijing"
	HongKong            = "oss-cn-hongkong"
	ShenZhen            = "oss-cn-shenzhen"
	ShangHai            = "oss-cn-shanghai"
	UsWest              = "oss-us-west-1"
	ApSouthEast         = "oss-ap-southeast-1"
	HangZhouInternal    = "oss-cn-hangzhou-internal"
	QingDaoInternal     = "oss-cn-qingdao-internal"
	BeiJingInternal     = "oss-cn-beijing-internal"
	HongKongInternal    = "oss-cn-hongkong-internal"
	ShenZhenInternal    = "oss-cn-shenzhen-internal"
	ShangHaiInternal    = "oss-cn-shanghai-internal"
	UsWestInternal      = "oss-us-west-1-internal"
	ApSouthEastInternal = "oss-ap-southeast-1-internal"

	DefaultRegion = HangZhou
)

// The OSS type encapsulates operations with an OSS region.

type Auth struct {
	AccessKey string
	SecretKey string
}

type OSS struct {
	Auth
	Region string
}

// The Bucket type encapsulates operations with an OSS bucket.
type Bucket struct {
	*OSS
	Name string
}

// The Owner type represents the owner of the object in an OSS bucket.
type Owner struct {
	ID          string
	DisplayName string
}

var (
	attempts        = defaultAttempts
	defaultAttempts = AttemptStrategy{
		Min:   5,
		Total: 5 * time.Second,
		Delay: 200 * time.Millisecond,
	}
)

// RetryAttempts sets whether failing OSS requests may be retried to cope
// with eventual consistency or temporary failures. It should not be
// called while operations are in progress.
func RetryAttempts(retry bool) {
	if retry {
		attempts = defaultAttempts
	} else {
		attempts = AttemptStrategy{}
	}
}

// New creates a new OSS.
func New(region, accessId, accessKey string) *OSS {
	auth := Auth{accessId, accessKey}
	return &OSS{auth, region}
}

// Bucket returns a Bucket with the given name.
func (oss *OSS) Bucket(name string) *Bucket {
	return &Bucket{oss, name}
}

var createBucketConfiguration = `<CreateBucketConfiguration xmlns="http://doc.oss.aliyuncs.com"> 
  <LocationConstraint>%s</LocationConstraint> 
</CreateBucketConfiguration>`

// locationConstraint returns an io.Reader specifying a LocationConstraint if
// required for the region.
//
// See http://goo.gl/bh9Kq for details.
// func (OSS *OSS) locationConstraint() io.Reader {
// 	constraint := ""
// 	if OSS.Region.OSSLocationConstraint {
// 		constraint = fmt.Sprintf(createBucketConfiguration, OSS.Region.Name)
// 	}
// 	return strings.NewReader(constraint)
// }

type ACL string

const (
	Private           = ACL("private")
	PublicRead        = ACL("public-read")
	PublicReadWrite   = ACL("public-read-write")
	AuthenticatedRead = ACL("authenticated-read")
	BucketOwnerRead   = ACL("bucket-owner-read")
	BucketOwnerFull   = ACL("bucket-owner-full-control")
)

// PutBucket creates a new bucket.
//
// See http://goo.gl/ndjnR for details.
func (b *Bucket) PutBucket(perm ACL) error {
	headers := map[string][]string{
		"x-oss-acl": {string(perm)},
	}
	req := &request{
		method:  "PUT",
		bucket:  b.Name,
		path:    "/",
		headers: headers,
		// payload: b.locationConstraint(),
	}
	return b.OSS.query(req, nil)
}

// DelBucket removes an existing OSS bucket. All objects in the bucket must
// be removed before the bucket itself can be removed.
//
// See http://goo.gl/GoBrY for details.
func (b *Bucket) DelBucket() (err error) {
	req := &request{
		method: "DELETE",
		bucket: b.Name,
		path:   "/",
	}
	for attempt := attempts.Start(); attempt.Next(); {
		err = b.OSS.query(req, nil)
		if !shouldRetry(err) {
			break
		}
	}
	return err
}

// Get retrieves an object from an OSS bucket.
//
// See http://goo.gl/isCO7 for details.
func (b *Bucket) Get(path string) (data []byte, err error) {
	body, err := b.GetReader(path)
	if err != nil {
		return nil, err
	}
	data, err = ioutil.ReadAll(body)
	body.Close()
	return data, err
}

// GetReader retrieves an object from an OSS bucket.
// It is the caller's responsibility to call Close on rc when
// finished reading.
func (b *Bucket) GetReader(path string) (rc io.ReadCloser, err error) {
	req := &request{
		bucket: b.Name,
		path:   path,
	}
	err = b.OSS.prepare(req)
	if err != nil {
		return nil, err
	}
	for attempt := attempts.Start(); attempt.Next(); {
		hresp, err := b.OSS.run(req)
		if shouldRetry(err) && attempt.HasNext() {
			continue
		}
		if err != nil {
			return nil, err
		}
		return hresp.Body, nil
	}
	panic("unreachable")
}

// Put inserts an object into the OSS bucket.
//
// See http://goo.gl/FEBPD for details.
func (b *Bucket) Put(path string, data []byte, contType string, perm ACL) error {
	body := bytes.NewBuffer(data)
	return b.PutReader(path, body, int64(len(data)), contType, perm)
}

// PutReader inserts an object into the OSS bucket by consuming data
// from r until EOF.
func (b *Bucket) PutReader(path string, r io.Reader, length int64, contType string, perm ACL) error {
	headers := map[string][]string{
		"Content-Length": {strconv.FormatInt(length, 10)},
		"Content-Type":   {contType},
		"x-oss-acl":      {string(perm)},
	}
	req := &request{
		method:  "PUT",
		bucket:  b.Name,
		path:    path,
		headers: headers,
		payload: r,
	}
	return b.OSS.query(req, nil)
}

// Del removes an object from the OSS bucket.
//
// See http://goo.gl/APeTt for details.
func (b *Bucket) Del(path string) error {
	req := &request{
		method: "DELETE",
		bucket: b.Name,
		path:   path,
	}
	return b.OSS.query(req, nil)
}

// Head retrieves metadata from an object without returning the object itself.
// This operation is useful if you are interested only in an object's metadata.
func (b *Bucket) Head(path string) (resp *http.Response, err error) {
	req := &request{
		method: "HEAD",
		bucket: b.Name,
		path:   path,
	}
	resp, err = b.OSS.queryRaw(req)
	return
}

// The ListResp type holds the results of a List bucket operation.
type ListResp struct {
	Name       string
	Prefix     string
	Delimiter  string
	Marker     string
	MaxKeys    int
	NextMarker string
	// IsTruncated is true if the results have been truncated because
	// there are more keys and prefixes than can fit in MaxKeys.
	// N.B. this is the opposite sense to that documented (incorrectly) in
	// http://goo.gl/YjQTc
	IsTruncated    bool
	Contents       []ContentInfo
	CommonPrefixes []string `xml:">Prefix"`
}

// The ContentInfo type holds individual file's information.
type ContentInfo struct {
	Key          string
	LastModified string
	ETag         string
	Type         string
	Size         int64
	StorageClass string
	Owner        OwnerInfo
}

// The OwnerInfo type holds file's Owner information.
type OwnerInfo struct {
	ID          string
	DisplayName string
}

// The Key type represents an item stored in an OSS bucket.
type Key struct {
	Key          string
	LastModified string
	Size         int64
	// ETag gives the hex-encoded MD5 sum of the contents,
	// surrounded with double-quotes.
	ETag         string
	StorageClass string
	Owner        Owner
}

// List returns information about objects in an S3 bucket.
//
// The prefix parameter limits the response to keys that begin with the
// specified prefix.
//
// The delim parameter causes the response to group all of the keys that
// share a common prefix up to the next delimiter in a single entry within
// the CommonPrefixes field. You can use delimiters to separate a bucket
// into different groupings of keys, similar to how folders would work.
//
// The marker parameter specifies the key to start with when listing objects
// in a bucket. aliyun OSS lists objects in alphabetical order and
// will return keys alphabetically greater than the marker.
//
// The max parameter specifies how many keys + common prefixes to return in
// the response. The default is 100.
//
// For example, given these keys in a bucket:
//
//     index.html
//     index2.html
//     photos/2006/January/sample.jpg
//     photos/2006/February/sample2.jpg
//     photos/2006/February/sample3.jpg
//     photos/2006/February/sample4.jpg
//
// Listing this bucket with delimiter set to "/" would yield the
// following result:
//
//     &ListResp{
//         Name:      "sample-bucket",
//         MaxKeys:   100,
//         Delimiter: "/",
//         Contents:  []Key{
//             {Key: "index.html", "index2.html"},
//         },
//         CommonPrefixes: []string{
//             "photos/",
//         },
//     }
//
// Listing the same bucket with delimiter set to "/" and prefix set to
// "photos/2006/" would yield the following result:
//
//     &ListResp{
//         Name:      "sample-bucket",
//         MaxKeys:   100,
//         Delimiter: "/",
//         Prefix:    "photos/2006/",
//         CommonPrefixes: []string{
//             "photos/2006/February/",
//             "photos/2006/January/",
//         },
//     }
//
// See http://goo.gl/YjQTc for details.
func (b *Bucket) List(prefix, delim, marker string, max int) (result *ListResp, err error) {
	params := map[string][]string{
		"prefix":    {prefix},
		"delimiter": {delim},
		"marker":    {marker},
	}
	if max > 0 && max <= 1000 {
		params["max-keys"] = []string{strconv.FormatInt(int64(max), 10)}
	}
	req := &request{
		bucket: b.Name,
		params: params,
	}
	result = &ListResp{}
	for attempt := attempts.Start(); attempt.Next(); {
		err = b.OSS.query(req, result)
		if !shouldRetry(err) {
			break
		}
	}
	if err != nil {
		return nil, err
	}
	return result, nil
}

// URL returns a non-signed URL that allows retriving the
// object at path. It only works if the object is publicly
// readable (see SignedURL).
func (b *Bucket) URL(path string) string {
	req := &request{
		bucket: b.Name,
		path:   path,
	}
	err := b.OSS.prepare(req)
	if err != nil {
		panic(err)
	}
	u, err := req.url()
	if err != nil {
		panic(err)
	}
	u.RawQuery = ""
	return u.String()
}

// SignedURL returns a signed URL that allows anyone holding the URL
// to retrieve the object at path. The signature is valid until expires.
func (b *Bucket) SignedURL(path string, expires time.Time) string {
	req := &request{
		bucket: b.Name,
		path:   path,
		params: url.Values{"Expires": {strconv.FormatInt(expires.Unix(), 10)}},
	}
	err := b.OSS.prepare(req)
	if err != nil {
		panic(err)
	}
	u, err := req.url()
	if err != nil {
		panic(err)
	}
	return u.String()
}

type request struct {
	method   string
	bucket   string
	path     string
	signpath string
	params   url.Values
	headers  http.Header
	baseurl  string
	payload  io.Reader
	prepared bool
}

func (req *request) url() (*url.URL, error) {
	u, err := url.Parse(req.baseurl)
	if err != nil {
		return nil, fmt.Errorf("bad OSS endpoint URL %q: %v", req.baseurl, err)
	}
	u.RawQuery = req.params.Encode()
	u.Path = req.path
	return u, nil
}

// query prepares and runs the req request.
// If resp is not nil, the XML data contained in the response
// body will be unmarshalled on it.
func (oss *OSS) query(req *request, resp interface{}) error {
	err := oss.prepare(req)
	if err != nil {
		return err
	}
	hresp, err := oss.run(req)
	if err != nil {
		return err
	}
	if resp != nil {
		err = xml.NewDecoder(hresp.Body).Decode(resp)
	}
	hresp.Body.Close()
	return nil
}

// queryRaw prepares and runs the req request but returns raw response instead of decode it.
// It's the caller's duty to call hresp.Body.Close() after handled http response.
func (oss *OSS) queryRaw(req *request) (*http.Response, error) {
	err := oss.prepare(req)
	if err != nil {
		return nil, err
	}
	hresp, err := oss.run(req)
	return hresp, err
}

// prepare sets up req to be delivered to OSS.
func (oss *OSS) prepare(req *request) error {
	if !req.prepared {
		req.prepared = true
		if req.method == "" {
			req.method = "GET"
		}
		// Copy so they can be mutated without affecting on retries.
		params := make(url.Values)
		headers := make(http.Header)
		for k, v := range req.params {
			params[k] = v
		}
		for k, v := range req.headers {
			headers[k] = v
		}
		req.params = params
		req.headers = headers
		if !strings.HasPrefix(req.path, "/") {
			req.path = "/" + req.path
		}
		req.signpath = req.path

		req.baseurl = DefaultHost
		if oss.Region != "" {
			//			req.baseurl = fmt.Sprintf("http://%s.aliyuncs.com", oss.Region)
			req.baseurl = fmt.Sprintf("http://%s.%s.aliyuncs.com", req.bucket, oss.Region)
		}

		if req.bucket != "" {
			// Just in case, prevent injection.
			if strings.IndexAny(req.bucket, "/:@") >= 0 {
				return fmt.Errorf("bad oss bucket: %q", req.bucket)
			}
			//			req.path = "/" + req.bucket + req.path
			req.signpath = "/" + req.bucket + req.signpath
		}
	}

	// Always sign again as it's not clear how far the
	// server has handled a previous attempt.
	u, err := url.Parse(req.baseurl)
	if err != nil {
		return fmt.Errorf("bad oss endpoint URL %q: %v", req.baseurl, err)
	}
	req.headers["Host"] = []string{u.Host}
	req.headers["Date"] = []string{time.Now().In(time.UTC).Format(http.TimeFormat)}
	sign(oss.Auth, req.method, req.signpath, req.params, req.headers)
	return nil
}

// run sends req and returns the http response from the server.
func (oss *OSS) run(req *request) (*http.Response, error) {
	if debug {
		log.Printf("Running OSS request: %#v", req)
	}

	u, err := req.url()
	if err != nil {
		return nil, err
	}

	hreq := http.Request{
		URL:        u,
		Method:     req.method,
		ProtoMajor: 1,
		ProtoMinor: 1,
		Close:      true,
		Header:     req.headers,
	}

	if v, ok := req.headers["Content-Length"]; ok {
		hreq.ContentLength, _ = strconv.ParseInt(v[0], 10, 64)
		delete(req.headers, "Content-Length")
	}

	if req.payload != nil {
		hreq.Body = ioutil.NopCloser(req.payload)
	}

	hresp, err := http.DefaultClient.Do(&hreq)
	if err != nil {
		return nil, err
	}
	if debug {
		dump, _ := httputil.DumpResponse(hresp, true)
		log.Printf("} -> %s\n", dump)
	}
	if hresp.StatusCode != 200 && hresp.StatusCode != 204 {
		return nil, buildError(hresp)
	}
	return hresp, err
}

// Error represents an error in an operation with OSS.
type Error struct {
	StatusCode int    // HTTP status code (200, 403, ...)
	Code       string // EC2 error code ("UnsupportedOperation", ...)
	Message    string // The human-oriented error message
	BucketName string
	RequestId  string
	HostId     string
}

func (e *Error) Error() string {
	return e.Message
}

func buildError(r *http.Response) error {
	if debug {
		log.Printf("got error (status code %v)", r.StatusCode)
		data, err := ioutil.ReadAll(r.Body)
		if err != nil {
			log.Printf("\tread error: %v", err)
		} else {
			log.Printf("\tdata:\n%s\n\n", data)
		}
		r.Body = ioutil.NopCloser(bytes.NewBuffer(data))
	}

	err := Error{}
	// TODO return error if Unmarshal fails?
	xml.NewDecoder(r.Body).Decode(&err)
	r.Body.Close()
	err.StatusCode = r.StatusCode
	if err.Message == "" {
		err.Message = r.Status
	}
	if debug {
		log.Printf("err: %#v\n", err)
	}
	return &err
}

func shouldRetry(err error) bool {
	if err == nil {
		return false
	}
	switch err {
	case io.ErrUnexpectedEOF, io.EOF:
		return true
	}
	switch e := err.(type) {
	case *net.DNSError:
		return true
	case *net.OpError:
		switch e.Op {
		case "read", "write":
			return true
		}
	case *Error:
		switch e.Code {
		case "InternalError", "NoSuchUpload", "NoSuchBucket":
			return true
		}
	}
	return false
}

func hasCode(err error, code string) bool {
	osserr, ok := err.(*Error)
	return ok && osserr.Code == code
}