Put takes the body's exact size and always sends it as Content-Length; refuse a link under one second (issue 1592)

go/objectlink/objectlink.go

@@ -47,7 +47,7 @@ type Request struct {
 	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 is how long the link lives: at least one second, at most ten minutes.
 	Expires time.Duration
 	// Now is the signing time; the clock's when zero.
 	Now time.Time
@@ -58,8 +58,8 @@ 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 {
+	if r.Expires < time.Second || 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)
+		return "", fmt.Errorf("objectlink: a link lives at least 1s 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")
@@ -171,66 +171,48 @@ 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.
+// exact counts and hashes what passes, and fails a stream that is longer than declared, or ends before it.
-type limited struct {
+type exact struct {
-	r     io.Reader
+	r    io.Reader
-	limit int64
+	size int64
-	n     int64
+	n    int64
-	hash  interface{ Write([]byte) (int, error) }
+	hash interface{ Write([]byte) (int, error) }
 }
 
-func (l *limited) Read(p []byte) (int, error) {
+func (l *exact) Read(p []byte) (int, error) {
 	n, err := l.r.Read(p)
 	l.n += int64(n)
-	if l.n > l.limit {
+	if l.n > l.size {
-		return 0, tooBig(l.limit)
+		return 0, fmt.Errorf("objectlink: the body is longer than the declared %d bytes", l.size)
 	}
 	l.hash.Write(p[:n])
+	if err == io.EOF && l.n < l.size {
+		return n, fmt.Errorf("objectlink: the body ended at %d bytes, short of the declared %d", l.n, l.size)
+	}
 	return n, err
 }
 
-// sizeOf is the body's length where it is known without reading it, else -1.
+// Put streams body to a link made for PUT, never holding it in memory. The caller gives the body's exact size, which
-func sizeOf(body io.Reader) int64 {
+// is always sent as the Content-Length (a store answers 411 to a PUT without one). A negative size is refused, and so
-	switch b := body.(type) {
+// is one over limit (MaxDocument when limit is not above zero, ADR 0468 §3), both before any request. A body that
-	case interface{ Len() int }:
+// yields more or fewer bytes than size fails the request, so a PUT whose body does not end as declared stores nothing.
-		return int64(b.Len())
+func Put(ctx context.Context, client *http.Client, link string, body io.Reader, size, limit int64) (Result, error) {
-	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 < 0 {
+		return Result{}, errors.New("objectlink: put needs the size of the body; a negative or unknown size is refused")
+	}
 	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}
+	counted := &exact{r: body, size: size, hash: hash}
 	req, err := http.NewRequestWithContext(ctx, http.MethodPut, link, io.NopCloser(counted))
 	if err != nil {
 		return Result{}, err
@@ -242,15 +224,15 @@ func Put(ctx context.Context, client *http.Client, link string, body io.Reader,
 	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)
 	}
+	if counted.n != size {
+		return Result{}, fmt.Errorf("objectlink: put: sent %d bytes of the declared %d", counted.n, size)
+	}
 	return Result{Size: counted.n, SHA256: hex.EncodeToString(hash.Sum(nil))}, nil
 }
 

go/objectlink/objectlink_test.go

@@ -52,7 +52,7 @@ func TestPresignRefusesBadExpiresAndKeepsTheHost(t *testing.T) {
 		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} {
+	for _, e := range []time.Duration{0, -time.Second, 500 * time.Millisecond, 601 * time.Second, time.Hour} {
 		r.Expires = e
 		if _, err := Presign(r); err == nil {
 			t.Errorf("Expires %s was accepted", e)
@@ -60,54 +60,71 @@ func TestPresignRefusesBadExpiresAndKeepsTheHost(t *testing.T) {
 	}
 }
 
-// 1592/AC3: Put streams and answers size and SHA-256 as the server saw them; past the limit it is refused;
+// 1592/AC3: Put streams and answers size and SHA-256 as the server saw them, always with the declared Content-Length;
-// a 403 with an S3 error names its code.
+// an unknown size, one past the limit, or a body not ending as declared is refused; a 403 with an S3 error names its code.
 func TestPutStreamsCountsAndRefuses(t *testing.T) {
 	var gotBody []byte
-	var gotLength int64
+	var lengths []int64
 	store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		gotBody, _ = io.ReadAll(r.Body)
-		gotLength = r.ContentLength
+		lengths = append(lengths, 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.
+	// A reader with no Len or Seek: still sent with the declared length.
-	got, err := Put(context.Background(), store.Client(), store.URL+"/b/k", io.MultiReader(bytes.NewReader(data)), 0)
+	got, err := Put(context.Background(), store.Client(), store.URL+"/b/k", io.MultiReader(bytes.NewReader(data)), 1000, 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, 1000); err != nil {
-	if _, err := Put(context.Background(), store.Client(), store.URL+"/b/k", bytes.NewReader(data), 1000); err != nil || gotLength != 1000 {
+		t.Errorf("exactly the limit: %v", err)
-		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.
+	for _, l := range lengths {
-	reads := &http.Client{Transport: roundTrip(func(r *http.Request) (*http.Response, error) {
+		if l != 1000 {
-		_, err := io.Copy(io.Discard, r.Body)
+			t.Errorf("a request carried Content-Length %d, not the declared 1000", l)
-		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)
+	if len(lengths) != 2 {
-	var refused *StoreError
+		t.Fatalf("the server saw %d requests", len(lengths))
-	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)
+	// Refused before any request: unknown (negative) size, and a size past the limit.
+	if _, err := Put(context.Background(), store.Client(), store.URL+"/b/k", bytes.NewReader(data), -1, 0); err == nil || !strings.Contains(err.Error(), "size") {
+		t.Errorf("an unknown size: %v", err)
+	}
+	if _, err := Put(context.Background(), store.Client(), store.URL+"/b/k", bytes.NewReader(data), 1000, 999); err == nil || !strings.Contains(err.Error(), "ADR 0468 §3") {
+		t.Errorf("a size past the limit: %v", err)
+	}
+	if len(lengths) != 2 {
+		t.Errorf("a refused call made a request: the server saw %d", len(lengths))
+	}
+
+	// A body longer or shorter than declared fails, and the store is not told it succeeded.
+	reads := &http.Client{Transport: roundTrip(func(r *http.Request) (*http.Response, error) {
+		_, err := io.Copy(io.Discard, r.Body)
+		if err == nil {
+			return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(""))}, nil
+		}
+		return nil, err
+	})}
+	for name, declared := range map[string]int64{"longer than declared": 999, "shorter than declared": 1001} {
+		if _, err := Put(context.Background(), reads, store.URL+"/b/k", bytes.NewReader(data), declared, 0); err == nil {
+			t.Errorf("a body %s was accepted", name)
+		}
+	}
+
+	_, err = Put(context.Background(), store.Client(), store.URL+"/b/k?deny=1", bytes.NewReader(data), 1000, 0)
+	e, ok := err.(*StoreError)
+	if !ok || e.Status != 403 || e.Code != "SignatureDoesNotMatch" || !strings.Contains(e.Error(), "SignatureDoesNotMatch") {
+		t.Fatalf("the refusal is %v", err)
 	}
 }