minstrel/internal/backup/s3.go

package backup

import (
	"context"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

// ErrS3 is a request the bucket refused or never answered.
var ErrS3 = errors.New("backup: the S3 bucket refused the request")

// Bucket is a path-style S3 bucket: an endpoint such as Fabled Archives'
// backup port or MinIO, a bucket name, and a key pair. A backup only ever
// adds objects and lists them, so an append-only key is enough.
type Bucket struct {
	Endpoint  string
	Name      string
	AccessKey string
	SecretKey string
	// Region defaults to us-east-1, which S3-compatible servers accept.
	Region string
	Client *http.Client
	// now is the signing clock; tests pin it.
	now func() time.Time
}

// Object is one stored object.
type Object struct {
	Key          string
	Size         int64
	LastModified time.Time
}

func (b Bucket) region() string {
	if b.Region == "" {
		return "us-east-1"
	}
	return b.Region
}

func (b Bucket) client() *http.Client {
	if b.Client != nil {
		return b.Client
	}
	return http.DefaultClient
}

func (b Bucket) clock() time.Time {
	if b.now != nil {
		return b.now()
	}
	return time.Now()
}

// objectURL is the path-style URL of key, encoded exactly as it is signed.
func (b Bucket) objectURL(key string) (*url.URL, error) {
	u, err := url.Parse(strings.TrimRight(b.Endpoint, "/"))
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return nil, fmt.Errorf("%w: the endpoint %q is not an http(s) address", ErrS3, b.Endpoint)
	}
	if b.Name == "" {
		return nil, fmt.Errorf("%w: no bucket name", ErrS3)
	}
	raw := "/" + uriEncode(b.Name)
	if key != "" {
		for s := range strings.SplitSeq(key, "/") {
			raw += "/" + uriEncode(s)
		}
	}
	u.RawPath = u.EscapedPath() + raw
	u.Path, _ = url.PathUnescape(u.RawPath)
	return u, nil
}

func (b Bucket) do(req *http.Request, payloadSHA256 string) (*http.Response, error) {
	sign(req, payloadSHA256, b.AccessKey, b.SecretKey, b.region(), b.clock())
	resp, err := b.client().Do(req)
	if err != nil {
		return nil, fmt.Errorf("%w: %s is unreachable: %v", ErrS3, req.URL.Host, err)
	}
	if resp.StatusCode/100 != 2 {
		defer func() { _ = resp.Body.Close() }()
		return nil, refusal(resp)
	}
	return resp, nil
}

// Put stores body under key. size is the body's length and payloadSHA256
// its hex SHA-256, which the request is signed over.
func (b Bucket) Put(ctx context.Context, key string, body io.Reader, size int64, payloadSHA256 string) error {
	u, err := b.objectURL(key)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, u.String(), body)
	if err != nil {
		return err
	}
	req.URL = u
	req.ContentLength = size
	if size == 0 {
		req.Body = http.NoBody
	}
	resp, err := b.do(req, payloadSHA256)
	if err != nil {
		return err
	}
	drain(resp)
	return nil
}

// List returns up to limit objects whose keys start with prefix, in key
// order.
func (b Bucket) List(ctx context.Context, prefix string, limit int) ([]Object, error) {
	if limit < 1 {
		limit = 1000
	}
	var out []Object
	token := ""
	for {
		u, err := b.objectURL("")
		if err != nil {
			return nil, err
		}
		q := url.Values{"list-type": {"2"}, "prefix": {prefix}}
		if page := limit - len(out); page < 1000 {
			q.Set("max-keys", strconv.Itoa(page))
		}
		if token != "" {
			q.Set("continuation-token", token)
		}
		u.RawQuery = canonicalQuery(q)
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
		if err != nil {
			return nil, err
		}
		req.URL = u
		resp, err := b.do(req, emptySHA256)
		if err != nil {
			return nil, err
		}
		var page struct {
			Contents []struct {
				Key          string `xml:"Key"`
				Size         int64  `xml:"Size"`
				LastModified string `xml:"LastModified"`
			} `xml:"Contents"`
			IsTruncated           bool   `xml:"IsTruncated"`
			NextContinuationToken string `xml:"NextContinuationToken"`
		}
		err = xml.NewDecoder(io.LimitReader(resp.Body, 16<<20)).Decode(&page)
		drain(resp)
		if err != nil {
			return nil, fmt.Errorf("%w: the listing is not S3 XML: %v", ErrS3, err)
		}
		for _, c := range page.Contents {
			modified, _ := time.Parse(time.RFC3339, c.LastModified)
			out = append(out, Object{Key: c.Key, Size: c.Size, LastModified: modified})
		}
		if !page.IsTruncated || page.NextContinuationToken == "" || len(out) >= limit {
			return out, nil
		}
		token = page.NextContinuationToken
	}
}

// refusal turns an S3 error response into words an admin can act on.
func refusal(resp *http.Response) error {
	var e struct {
		Code    string `xml:"Code"`
		Message string `xml:"Message"`
	}
	_ = xml.NewDecoder(io.LimitReader(resp.Body, 64<<10)).Decode(&e)
	hint := map[string]string{
		"SignatureDoesNotMatch": "the secret key is wrong",
		"InvalidAccessKeyId":    "the access key is not known to this server",
		"AccessDenied":          "the key may not do this in this bucket",
		"NoSuchBucket":          "there is no bucket by that name",
		"SlowDown":              "the server asks to slow down",
	}[e.Code]
	switch {
	case hint != "":
		return fmt.Errorf("%w: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, hint)
	case e.Code != "":
		return fmt.Errorf("%w: HTTP %d %s: %s", ErrS3, resp.StatusCode, e.Code, e.Message)
	default:
		return fmt.Errorf("%w: HTTP %d", ErrS3, resp.StatusCode)
	}
}

func drain(resp *http.Response) {
	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20))
	_ = resp.Body.Close()
}