package s3
import (
"bytes"
"context"
"encoding/xml"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
)
// What unversioned files (RFC 0020) need beyond what attachments do:
// objects whose metadata is part of the record, read back one by one,
// and a listing of a prefix. The bucket is the record there and the
// forge's catalog a copy rebuilt from it, so the forge has to be able to
// read everything it wrote, including the fossil mtime and hash it wrote
// beside the bytes.
// metaPrefix is how S3 spells user metadata on the wire.
const metaPrefix = "x-amz-meta-"
// Object is one object as a HEAD or a listing reports it. Meta is the
// user metadata by its bare, lower-cased name ("mtime", not
// "x-amz-meta-mtime"); a listing does not carry it.
type Object struct {
Key string
Size int64
Meta map[string]string
}
// Object headers a PUT may set beyond the body's length.
type PutOptions struct {
ContentType string
// Disposition is stored with the object and served with it, on a
// presigned GET and from a public base alike; empty stores none.
Disposition string
// Meta is user metadata by bare name; S3 lower-cases it.
Meta map[string]string
}
// PutWith is Put with the object's own headers. Every one of them is a
// signed header, so the values are the forge's and the request must
// carry exactly them. Values must be US-ASCII, which is S3's rule for
// headers on a signed request; the caller encodes anything else.
func (b Bucket) PutWith(ctx context.Context, client *http.Client, key string, body []byte, opt PutOptions, now time.Time) error {
if client == nil {
client = http.DefaultClient
}
headers := map[string]string{
"content-length": fmt.Sprint(len(body)),
"content-type": opt.ContentType,
}
if opt.Disposition != "" {
headers["content-disposition"] = opt.Disposition
}
for k, v := range opt.Meta {
headers[metaPrefix+strings.ToLower(k)] = v
}
put, err := b.Presign(http.MethodPut, key, 5*time.Minute, now, headers)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, put.String(), bytes.NewReader(body))
if err != nil {
return err
}
req.ContentLength = int64(len(body))
for k, v := range headers {
if k != "content-length" {
req.Header.Set(k, v)
}
}
return do(client, req)
}
// Stat asks whether key is in the bucket, how big it is and what
// metadata it carries. A 404 is the answer false, not an error.
func (b Bucket) Stat(ctx context.Context, client *http.Client, key string, now time.Time) (Object, bool, error) {
if client == nil {
client = http.DefaultClient
}
u, err := b.Presign(http.MethodHead, key, 5*time.Minute, now, nil)
if err != nil {
return Object{}, false, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodHead, u.String(), nil)
if err != nil {
return Object{}, false, err
}
resp, err := client.Do(req)
if err != nil {
return Object{}, false, fmt.Errorf("HEAD: %w", redact(err, req.URL))
}
defer resp.Body.Close()
io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16))
switch {
case resp.StatusCode == http.StatusNotFound:
return Object{}, false, nil
case resp.StatusCode >= 200 && resp.StatusCode < 300:
o := Object{Key: key, Size: resp.ContentLength, Meta: map[string]string{}}
for name := range resp.Header {
if lower := strings.ToLower(name); strings.HasPrefix(lower, metaPrefix) {
o.Meta[strings.TrimPrefix(lower, metaPrefix)] = resp.Header.Get(name)
}
}
return o, true, nil
}
return Object{}, false, &ProbeError{Verb: http.MethodHead, Status: resp.StatusCode}
}
// maxListPages bounds one listing. At S3's thousand keys a page this is
// ten million objects, which no repository's unversioned files come
// near; the bound is there so a bucket that hands back the same token
// forever cannot hold the forge in a loop.
const maxListPages = 10000
// List is every object under prefix, by ListObjectsV2, following the
// continuation token to the end. Keys come back in the bucket's order,
// which S3 defines as byte order.
func (b Bucket) List(ctx context.Context, client *http.Client, prefix string, now time.Time) ([]Object, error) {
if client == nil {
client = http.DefaultClient
}
var out []Object
token := ""
for range maxListPages {
extra := url.Values{}
extra.Set("list-type", "2")
extra.Set("prefix", prefix)
// Keys come back URL-encoded. XML 1.0 cannot carry most control
// characters at all, so without this a key holding one makes
// the whole page unreadable.
extra.Set("encoding-type", "url")
if token != "" {
extra.Set("continuation-token", token)
}
u, err := b.presignAt(http.MethodGet, "", 5*time.Minute, now, nil, extra)
if err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
page, err := listPage(client, req)
if err != nil {
return nil, err
}
for _, c := range page.Contents {
key := c.Key
// Decoded only when the answer says it was encoded: a
// service that ignores encoding-type sends keys as they
// are, and decoding one of those would turn every "+" in
// it into a space.
if page.EncodingType == "url" {
var err error
if key, err = url.QueryUnescape(c.Key); err != nil {
return nil, fmt.Errorf("s3: listing %s: key %q is not URL-encoded: %w", prefix, c.Key, err)
}
}
out = append(out, Object{Key: key, Size: c.Size})
}
if !page.IsTruncated || page.NextContinuationToken == "" {
return out, nil
}
token = page.NextContinuationToken
}
return nil, fmt.Errorf("s3: listing %s did not end after %d pages", prefix, maxListPages)
}
// listResult is the part of a ListObjectsV2 answer a listing reads.
// The tags carry no namespace, so they match S3's whatever it declares.
type listResult struct {
EncodingType string `xml:"EncodingType"`
IsTruncated bool `xml:"IsTruncated"`
NextContinuationToken string `xml:"NextContinuationToken"`
Contents []struct {
Key string `xml:"Key"`
Size int64 `xml:"Size"`
} `xml:"Contents"`
}
// maxListPage bounds one page's body. A thousand keys of a few hundred
// bytes each is well under it.
const maxListPage = 8 << 20
func listPage(client *http.Client, req *http.Request) (listResult, error) {
resp, err := client.Do(req)
if err != nil {
return listResult{}, fmt.Errorf("list: %w", redact(err, req.URL))
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, maxListPage+1))
if err != nil {
return listResult{}, err
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
pe := &ProbeError{Verb: "list", Status: resp.StatusCode}
if m := codeRE.FindSubmatch(body); m != nil {
pe.Code = string(m[1])
}
return listResult{}, pe
}
if len(body) > maxListPage {
return listResult{}, fmt.Errorf("s3: a listing page larger than %d bytes", maxListPage)
}
var page listResult
if err := xml.Unmarshal(body, &page); err != nil {
return listResult{}, fmt.Errorf("s3: reading a listing: %w", err)
}
return page, nil
}
// Copy writes dst from src inside the bucket by CopyObject, with the
// headers and metadata in opt replacing the source's. Copying a key to
// itself is how an object's metadata changes without its bytes moving:
// a fossil mtime-only update to an unversioned file.
func (b Bucket) Copy(ctx context.Context, client *http.Client, src, dst string, opt PutOptions, now time.Time) error {
if client == nil {
client = http.DefaultClient
}
headers := map[string]string{
"content-type": opt.ContentType,
// The source is URL-encoded as S3 requires, slashes kept.
"x-amz-copy-source": "/" + uriEncode(b.Name, false) + "/" + uriEncode(src, false),
"x-amz-metadata-directive": "REPLACE",
}
if opt.Disposition != "" {
headers["content-disposition"] = opt.Disposition
}
for k, v := range opt.Meta {
headers[metaPrefix+strings.ToLower(k)] = v
}
put, err := b.Presign(http.MethodPut, dst, 5*time.Minute, now, headers)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, put.String(), http.NoBody)
if err != nil {
return err
}
for k, v := range headers {
req.Header.Set(k, v)
}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("copy: %w", redact(err, req.URL))
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
if err != nil {
return fmt.Errorf("copy: %w", err)
}
// S3 can answer a copy 200 and report its failure in the body, since
// the status goes out before the copy is done; only a
// CopyObjectResult is a success.
if resp.StatusCode >= 200 && resp.StatusCode < 300 && !bytes.Contains(body, []byte("<Error>")) {
return nil
}
pe := &ProbeError{Verb: "copy", Status: resp.StatusCode}
if m := codeRE.FindSubmatch(body); m != nil {
pe.Code = string(m[1])
}
return pe
}
// GetMax reads an object whole when it is at most max bytes. Get's
// bound is for the forge's own small objects; this one is the caller's,
// so an unversioned file can be read up to its binding's limit and no
// further.
func (b Bucket) GetMax(ctx context.Context, client *http.Client, key string, max int64, now time.Time) ([]byte, error) {
if client == nil {
client = http.DefaultClient
}
u, err := b.Presign(http.MethodGet, key, 5*time.Minute, now, nil)
if err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("GET: %w", redact(err, req.URL))
}
defer resp.Body.Close()
if resp.StatusCode/100 != 2 {
return nil, &ProbeError{Verb: http.MethodGet, Status: resp.StatusCode}
}
if resp.ContentLength > max {
return nil, fmt.Errorf("s3: GET %s: object larger than %d bytes", key, max)
}
raw, err := io.ReadAll(io.LimitReader(resp.Body, max+1))
if err != nil {
return nil, err
}
if int64(len(raw)) > max {
return nil, fmt.Errorf("s3: GET %s: object larger than %d bytes", key, max)
}
return raw, nil
}