Add objectlink: one helper signs a short store link and streams a document over it with a size limit and a digest (issue 1592)
This commit is contained in:
@@ -0,0 +1,295 @@
|
||||
// Package objectlink signs and streams a link to one object in the mesh's object store (novox/hq ADR 0468 §3).
|
||||
//
|
||||
// Bytes move over HTTP between a party and the store, on a signed link of at most ten minutes for one object and one
|
||||
// method, signed for the name the `s3-bucket` binding gives; a document over 100 MiB is refused. Presign makes the
|
||||
// link (Signature Version 4, query form, path style, payload unsigned, `host` the only signed header); Put and Get
|
||||
// move the bytes without holding them in memory and count and hash them as they go.
|
||||
//
|
||||
// Standard library only: the SDK requires nothing of the mesh's forge.
|
||||
package objectlink
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
// MaxDocument is the largest document a link moves (ADR 0468 §3): the default limit of Put and Get.
|
||||
MaxDocument int64 = 100 << 20
|
||||
// MaxExpires is the longest a link lives (ADR 0468 §3).
|
||||
MaxExpires = 10 * time.Minute
|
||||
|
||||
defaultRegion = "us-east-1"
|
||||
algorithm = "AWS4-HMAC-SHA256"
|
||||
)
|
||||
|
||||
// Request is one link to sign.
|
||||
type Request struct {
|
||||
// Method is GET or PUT.
|
||||
Method string
|
||||
// Endpoint is the store as the party reaches it, "http://host:port". Its host is signed exactly as written.
|
||||
Endpoint string
|
||||
Bucket string
|
||||
Key string
|
||||
// AccessKey and SecretKey are the credential the s3-bucket binding gives.
|
||||
AccessKey string
|
||||
SecretKey string
|
||||
// Region defaults to us-east-1.
|
||||
Region string
|
||||
// Expires is how long the link lives: more than zero, at most ten minutes.
|
||||
Expires time.Duration
|
||||
// Now is the signing time; the clock's when zero.
|
||||
Now time.Time
|
||||
}
|
||||
|
||||
// Presign returns the signed link for one object and one method.
|
||||
func Presign(r Request) (string, error) {
|
||||
if r.Method != http.MethodGet && r.Method != http.MethodPut {
|
||||
return "", fmt.Errorf("objectlink: method %q: a link is for GET or PUT (ADR 0468 §3)", r.Method)
|
||||
}
|
||||
if r.Expires <= 0 || r.Expires > MaxExpires {
|
||||
return "", fmt.Errorf("objectlink: a link lives more than 0 and at most %s, not %s (ADR 0468 §3)", MaxExpires, r.Expires)
|
||||
}
|
||||
if r.Bucket == "" || r.Key == "" || r.AccessKey == "" || r.SecretKey == "" {
|
||||
return "", errors.New("objectlink: a bucket, a key and a credential are needed to sign a link")
|
||||
}
|
||||
u, err := url.Parse(r.Endpoint)
|
||||
if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
|
||||
return "", fmt.Errorf("objectlink: endpoint %q is not http(s)://host[:port]", r.Endpoint)
|
||||
}
|
||||
region := r.Region
|
||||
if region == "" {
|
||||
region = defaultRegion
|
||||
}
|
||||
now := r.Now
|
||||
if now.IsZero() {
|
||||
now = time.Now()
|
||||
}
|
||||
path := "/" + encode(r.Bucket, false) + "/" + encode(r.Key, false)
|
||||
query := sign(r.Method, u.Host, path, r.AccessKey, r.SecretKey, region, now.UTC(), int64(r.Expires/time.Second))
|
||||
return u.Scheme + "://" + u.Host + path + "?" + query, nil
|
||||
}
|
||||
|
||||
// sign returns the query string of a presigned link, signature last. The path is already encoded.
|
||||
func sign(method, host, path, accessKey, secretKey, region string, now time.Time, expires int64) string {
|
||||
date := now.Format("20060102")
|
||||
stamp := now.Format("20060102T150405Z")
|
||||
scope := date + "/" + region + "/s3/aws4_request"
|
||||
params := map[string]string{
|
||||
"X-Amz-Algorithm": algorithm,
|
||||
"X-Amz-Credential": accessKey + "/" + scope,
|
||||
"X-Amz-Date": stamp,
|
||||
"X-Amz-Expires": fmt.Sprint(expires),
|
||||
"X-Amz-SignedHeaders": "host",
|
||||
}
|
||||
names := make([]string, 0, len(params))
|
||||
for k := range params {
|
||||
names = append(names, k)
|
||||
}
|
||||
sort.Strings(names)
|
||||
var query []string
|
||||
for _, k := range names {
|
||||
query = append(query, encode(k, true)+"="+encode(params[k], true))
|
||||
}
|
||||
canonicalQuery := strings.Join(query, "&")
|
||||
canonical := strings.Join([]string{method, path, canonicalQuery, "host:" + host + "\n", "host", "UNSIGNED-PAYLOAD"}, "\n")
|
||||
digest := sha256.Sum256([]byte(canonical))
|
||||
toSign := strings.Join([]string{algorithm, stamp, scope, hex.EncodeToString(digest[:])}, "\n")
|
||||
key := mac([]byte("AWS4"+secretKey), date)
|
||||
key = mac(key, region)
|
||||
key = mac(key, "s3")
|
||||
key = mac(key, "aws4_request")
|
||||
return canonicalQuery + "&X-Amz-Signature=" + hex.EncodeToString(mac(key, toSign))
|
||||
}
|
||||
|
||||
func mac(key []byte, data string) []byte {
|
||||
h := hmac.New(sha256.New, key)
|
||||
h.Write([]byte(data))
|
||||
return h.Sum(nil)
|
||||
}
|
||||
|
||||
// encode is AWS's URI encoding: everything but A-Z a-z 0-9 - _ . ~ as %XX, and the slash kept in a path.
|
||||
func encode(s string, slash bool) string {
|
||||
var b strings.Builder
|
||||
for i := 0; i < len(s); i++ {
|
||||
c := s[i]
|
||||
switch {
|
||||
case c >= 'A' && c <= 'Z', c >= 'a' && c <= 'z', c >= '0' && c <= '9', c == '-', c == '_', c == '.', c == '~':
|
||||
b.WriteByte(c)
|
||||
case c == '/' && !slash:
|
||||
b.WriteByte(c)
|
||||
default:
|
||||
fmt.Fprintf(&b, "%%%02X", c)
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// Result is what moved: its size and the SHA-256 of its bytes, in hex.
|
||||
type Result struct {
|
||||
Size int64
|
||||
SHA256 string
|
||||
}
|
||||
|
||||
// StoreError is a refusal by the store: its status and the code and message of its XML error body.
|
||||
type StoreError struct {
|
||||
Status int
|
||||
Code string
|
||||
Message string
|
||||
}
|
||||
|
||||
func (e *StoreError) Error() string {
|
||||
if e.Code == "" {
|
||||
return fmt.Sprintf("objectlink: the store answered %d", e.Status)
|
||||
}
|
||||
return fmt.Sprintf("objectlink: the store answered %d %s: %s", e.Status, e.Code, e.Message)
|
||||
}
|
||||
|
||||
func storeError(resp *http.Response) error {
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
|
||||
var parsed struct {
|
||||
Code string `xml:"Code"`
|
||||
Message string `xml:"Message"`
|
||||
}
|
||||
_ = xml.Unmarshal(body, &parsed)
|
||||
return &StoreError{Status: resp.StatusCode, Code: parsed.Code, Message: parsed.Message}
|
||||
}
|
||||
|
||||
// tooBig is the error of a document past its limit.
|
||||
func tooBig(limit int64) error {
|
||||
return fmt.Errorf("objectlink: the document is over %d bytes (ADR 0468 §3)", limit)
|
||||
}
|
||||
|
||||
// limited counts and hashes what passes and fails the read that would pass the limit.
|
||||
type limited struct {
|
||||
r io.Reader
|
||||
limit int64
|
||||
n int64
|
||||
hash interface{ Write([]byte) (int, error) }
|
||||
}
|
||||
|
||||
func (l *limited) Read(p []byte) (int, error) {
|
||||
n, err := l.r.Read(p)
|
||||
l.n += int64(n)
|
||||
if l.n > l.limit {
|
||||
return 0, tooBig(l.limit)
|
||||
}
|
||||
l.hash.Write(p[:n])
|
||||
return n, err
|
||||
}
|
||||
|
||||
// sizeOf is the body's length where it is known without reading it, else -1.
|
||||
func sizeOf(body io.Reader) int64 {
|
||||
switch b := body.(type) {
|
||||
case interface{ Len() int }:
|
||||
return int64(b.Len())
|
||||
case io.Seeker:
|
||||
cur, err := b.Seek(0, io.SeekCurrent)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
end, err := b.Seek(0, io.SeekEnd)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
if _, err := b.Seek(cur, io.SeekStart); err != nil {
|
||||
return -1
|
||||
}
|
||||
return end - cur
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
// Put streams body to a link made for PUT. It never holds the body in memory, and refuses more than limit bytes
|
||||
// (MaxDocument when limit is not above zero): a length known beforehand is refused before anything is sent, and one
|
||||
// that is not is refused when the stream passes the limit. A body whose length is known (a Len method, or a Seeker) is
|
||||
// sent with it, as stores want; any other is sent chunked.
|
||||
func Put(ctx context.Context, client *http.Client, link string, body io.Reader, limit int64) (Result, error) {
|
||||
if limit <= 0 {
|
||||
limit = MaxDocument
|
||||
}
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
size := sizeOf(body)
|
||||
if size > limit {
|
||||
return Result{}, tooBig(limit)
|
||||
}
|
||||
hash := sha256.New()
|
||||
// Cancelled on the way out, so a refused stream tears its connection down and the store does not wait for the rest.
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
counted := &limited{r: body, limit: limit, hash: hash}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPut, link, io.NopCloser(counted))
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
req.ContentLength = size
|
||||
if size == 0 {
|
||||
req.Body = http.NoBody
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
cancel()
|
||||
if strings.Contains(err.Error(), tooBig(limit).Error()) {
|
||||
return Result{}, tooBig(limit)
|
||||
}
|
||||
return Result{}, fmt.Errorf("objectlink: put: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode > 299 {
|
||||
return Result{}, storeError(resp)
|
||||
}
|
||||
return Result{Size: counted.n, SHA256: hex.EncodeToString(hash.Sum(nil))}, nil
|
||||
}
|
||||
|
||||
// Get streams the object of a link made for GET into w, refusing more than limit bytes (MaxDocument when limit is
|
||||
// not above zero). When wantSHA256 is given and the bytes hash to another value the call fails: the caller removes
|
||||
// what it wrote.
|
||||
func Get(ctx context.Context, client *http.Client, link string, w io.Writer, limit int64, wantSHA256 string) (Result, error) {
|
||||
if limit <= 0 {
|
||||
limit = MaxDocument
|
||||
}
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, link, nil)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("objectlink: get: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode > 299 {
|
||||
return Result{}, storeError(resp)
|
||||
}
|
||||
if resp.ContentLength > limit {
|
||||
return Result{}, tooBig(limit)
|
||||
}
|
||||
hash := sha256.New()
|
||||
n, err := io.Copy(io.MultiWriter(w, hash), io.LimitReader(resp.Body, limit+1))
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("objectlink: get: %w", err)
|
||||
}
|
||||
if n > limit {
|
||||
return Result{}, tooBig(limit)
|
||||
}
|
||||
got := hex.EncodeToString(hash.Sum(nil))
|
||||
if wantSHA256 != "" && !strings.EqualFold(wantSHA256, got) {
|
||||
return Result{}, fmt.Errorf("objectlink: the document hashes to %s, not the %s expected", got, strings.ToLower(wantSHA256))
|
||||
}
|
||||
return Result{Size: n, SHA256: got}, nil
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
package objectlink
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Issue 1592: one helper signs and streams a store link (novox/hq ADR 0468 §3).
|
||||
|
||||
type roundTrip func(*http.Request) (*http.Response, error)
|
||||
|
||||
func (f roundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
|
||||
|
||||
func hexSum(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }
|
||||
|
||||
// 1592/AC1: AWS's documented query-presign example (a GET of test.txt, 86400 s, 2013-05-24) gives the documented signature.
|
||||
func TestPresignMatchesAWSDocumentedExample(t *testing.T) {
|
||||
now := time.Date(2013, 5, 24, 0, 0, 0, 0, time.UTC)
|
||||
query := sign("GET", "examplebucket.s3.amazonaws.com", "/test.txt", "AKIAIOSFODNN7EXAMPLE",
|
||||
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", "us-east-1", now, 86400)
|
||||
const want = "aeeed9bbccd4d02ee5c0109b86d86835f995330da4c265957d157751f604d404"
|
||||
if !strings.HasSuffix(query, "&X-Amz-Signature="+want) {
|
||||
t.Fatalf("the signature differs from AWS's documented one: %s", query)
|
||||
}
|
||||
if !strings.Contains(query, "X-Amz-Credential=AKIAIOSFODNN7EXAMPLE%2F20130524%2Fus-east-1%2Fs3%2Faws4_request") {
|
||||
t.Errorf("the credential is not encoded as documented: %s", query)
|
||||
}
|
||||
}
|
||||
|
||||
// 1592/AC2: an Expires of 0, below 0 or above 600 s is refused; the signed host is the endpoint's host.
|
||||
func TestPresignRefusesBadExpiresAndKeepsTheHost(t *testing.T) {
|
||||
r := Request{Method: "PUT", Endpoint: "http://minio.internal:9000", Bucket: "files", Key: "a b/c.pdf",
|
||||
AccessKey: "ak", SecretKey: "sk", Expires: 10 * time.Minute, Now: time.Unix(1700000000, 0)}
|
||||
link, err := Presign(r)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
u, err := url.Parse(link)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if u.Host != "minio.internal:9000" || u.Path != "/files/a b/c.pdf" || u.Query().Get("X-Amz-Expires") != "600" ||
|
||||
u.Query().Get("X-Amz-SignedHeaders") != "host" {
|
||||
t.Errorf("the link is %s", link)
|
||||
}
|
||||
for _, e := range []time.Duration{0, -time.Second, 601 * time.Second, time.Hour} {
|
||||
r.Expires = e
|
||||
if _, err := Presign(r); err == nil {
|
||||
t.Errorf("Expires %s was accepted", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1592/AC3: Put streams and answers size and SHA-256 as the server saw them; past the limit it is refused;
|
||||
// a 403 with an S3 error names its code.
|
||||
func TestPutStreamsCountsAndRefuses(t *testing.T) {
|
||||
var gotBody []byte
|
||||
var gotLength int64
|
||||
store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotBody, _ = io.ReadAll(r.Body)
|
||||
gotLength = r.ContentLength
|
||||
if r.URL.Query().Get("deny") != "" {
|
||||
w.WriteHeader(403)
|
||||
io.WriteString(w, `<?xml version="1.0"?><Error><Code>SignatureDoesNotMatch</Code><Message>nope</Message></Error>`)
|
||||
}
|
||||
}))
|
||||
defer store.Close()
|
||||
defer store.CloseClientConnections()
|
||||
data := bytes.Repeat([]byte("0123456789"), 100)
|
||||
|
||||
// A reader with no length: streamed chunked, counted as it goes.
|
||||
got, err := Put(context.Background(), store.Client(), store.URL+"/b/k", io.MultiReader(bytes.NewReader(data)), 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Size != int64(len(gotBody)) || got.SHA256 != hexSum(gotBody) || got.Size != 1000 {
|
||||
t.Errorf("answered %+v, the server received %d bytes %s", got, len(gotBody), hexSum(gotBody))
|
||||
}
|
||||
// One with a length is sent with it.
|
||||
if _, err := Put(context.Background(), store.Client(), store.URL+"/b/k", bytes.NewReader(data), 1000); err != nil || gotLength != 1000 {
|
||||
t.Errorf("exactly the limit: err %v, length %d", err, gotLength)
|
||||
}
|
||||
// Past the limit, both ways. A stand-in transport reads the body as a store would, so no connection is left half open.
|
||||
reads := &http.Client{Transport: roundTrip(func(r *http.Request) (*http.Response, error) {
|
||||
_, err := io.Copy(io.Discard, r.Body)
|
||||
return nil, err
|
||||
})}
|
||||
for name, body := range map[string]io.Reader{"unknown length": io.MultiReader(bytes.NewReader(data)), "known length": bytes.NewReader(data)} {
|
||||
if _, err := Put(context.Background(), reads, store.URL+"/b/k", body, 999); err == nil || !strings.Contains(err.Error(), "ADR 0468 §3") {
|
||||
t.Errorf("%s: one byte past the limit: %v", name, err)
|
||||
}
|
||||
}
|
||||
_, err = Put(context.Background(), store.Client(), store.URL+"/b/k?deny=1", bytes.NewReader(data), 0)
|
||||
var refused *StoreError
|
||||
if e, ok := err.(*StoreError); !ok || e.Status != 403 || e.Code != "SignatureDoesNotMatch" {
|
||||
t.Fatalf("the refusal is %v", err)
|
||||
} else {
|
||||
refused = e
|
||||
}
|
||||
if !strings.Contains(refused.Error(), "SignatureDoesNotMatch") {
|
||||
t.Errorf("the message does not name the code: %v", refused)
|
||||
}
|
||||
}
|
||||
|
||||
// 1592/AC4: Get writes the body and answers size and SHA-256; a wrong digest and a document past the limit are errors.
|
||||
func TestGetStreamsChecksAndRefuses(t *testing.T) {
|
||||
data := bytes.Repeat([]byte("abcdefghij"), 100)
|
||||
store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Write(data)
|
||||
}))
|
||||
defer store.Close()
|
||||
var out bytes.Buffer
|
||||
got, err := Get(context.Background(), store.Client(), store.URL+"/b/k", &out, 0, hexSum(data))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Size != 1000 || got.SHA256 != hexSum(data) || !bytes.Equal(out.Bytes(), data) {
|
||||
t.Errorf("answered %+v, wrote %d bytes", got, out.Len())
|
||||
}
|
||||
if _, err := Get(context.Background(), store.Client(), store.URL+"/b/k", io.Discard, 0, hexSum([]byte("other"))); err == nil {
|
||||
t.Error("a wrong digest was accepted")
|
||||
}
|
||||
if _, err := Get(context.Background(), store.Client(), store.URL+"/b/k", io.Discard, 999, ""); err == nil {
|
||||
t.Error("a document past the limit was accepted")
|
||||
}
|
||||
chunked := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.(http.Flusher).Flush()
|
||||
w.Write(data)
|
||||
}))
|
||||
defer chunked.Close()
|
||||
if _, err := Get(context.Background(), chunked.Client(), chunked.URL+"/b/k", io.Discard, 999, ""); err == nil {
|
||||
t.Error("a chunked document past the limit was accepted")
|
||||
}
|
||||
if _, err := Get(context.Background(), chunked.Client(), chunked.URL+"/b/k", io.Discard, 1000, ""); err != nil {
|
||||
t.Errorf("exactly the limit: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user