package api
import (
"bytes"
"context"
"crypto/sha1"
"encoding/base64"
"encoding/hex"
"errors"
"fmt"
"io"
"mime/multipart"
"net/http"
"net/http/httptest"
"net/url"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/uv"
)
// uvBucket is an S3 stand-in that keeps what unversioned files need:
// the bytes, the metadata the PUT signed, HEAD answering both, GET,
// a copy onto a key with the metadata replaced, DELETE,
// and a ListObjectsV2 that pages two keys at a time so a rebuild must
// follow the continuation token. Objects are kept per bucket name, so
// binding another bucket at the same endpoint sees none of them.
type uvBucket struct {
// raw makes the listing ignore encoding-type and send keys as they
// are, as a service that does not implement it would.
raw bool
// refusePut answers every PUT 500, as a bucket out of quota does.
refusePut bool
mu sync.Mutex
objects map[string]uvObject
lists int
gets int
}
type uvObject struct {
body []byte
meta map[string]string
ctype string
disposition string
}
func newUVBucket(t *testing.T) (*httptest.Server, *uvBucket) {
t.Helper()
b := &uvBucket{objects: map[string]uvObject{}}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
if q.Get("X-Amz-Signature") == "" {
w.WriteHeader(http.StatusForbidden)
return
}
b.mu.Lock()
defer b.mu.Unlock()
bucket, key, isObject := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/")
if !isObject {
b.lists++
var keys []string
for k := range b.objects {
if rest, ok := strings.CutPrefix(k, bucket+"/"); ok && strings.HasPrefix(rest, q.Get("prefix")) {
keys = append(keys, rest)
}
}
sort.Strings(keys)
start, _ := strconv.Atoi(q.Get("continuation-token"))
end := min(start+2, len(keys))
fmt.Fprint(w, "<ListBucketResult>")
encode := q.Get("encoding-type") == "url" && !b.raw
if encode {
fmt.Fprint(w, "<EncodingType>url</EncodingType>")
}
for _, k := range keys[start:end] {
shown := k
if encode {
shown = url.QueryEscape(k)
}
fmt.Fprintf(w, "<Contents><Key>%s</Key><Size>%d</Size></Contents>", shown, len(b.objects[bucket+"/"+k].body))
}
if end < len(keys) {
fmt.Fprintf(w, "<IsTruncated>true</IsTruncated><NextContinuationToken>%d</NextContinuationToken>", end)
}
fmt.Fprint(w, "</ListBucketResult>")
return
}
key = bucket + "/" + key
switch r.Method {
case http.MethodPut:
if b.refusePut {
w.WriteHeader(http.StatusInternalServerError)
return
}
meta := map[string]string{}
for name := range r.Header {
if lower := strings.ToLower(name); strings.HasPrefix(lower, "x-amz-meta-") {
meta[strings.TrimPrefix(lower, "x-amz-meta-")] = r.Header.Get(name)
}
}
body, _ := io.ReadAll(r.Body)
// CopyObject with REPLACE: the bytes from the source, the
// headers and metadata from this request.
if src := r.Header.Get("x-amz-copy-source"); src != "" {
from, err := url.PathUnescape(strings.TrimPrefix(src, "/"))
o, ok := b.objects[from]
if err != nil || !ok || r.Header.Get("x-amz-metadata-directive") != "REPLACE" {
w.WriteHeader(http.StatusNotFound)
return
}
body = o.body
}
b.objects[key] = uvObject{body: body, meta: meta, ctype: r.Header.Get("Content-Type"), disposition: r.Header.Get("Content-Disposition")}
case http.MethodHead:
o, ok := b.objects[key]
if !ok {
w.WriteHeader(http.StatusNotFound)
return
}
for k, v := range o.meta {
w.Header().Set("x-amz-meta-"+k, v)
}
w.Header().Set("Content-Length", strconv.Itoa(len(o.body)))
case http.MethodGet:
o, ok := b.objects[key]
if !ok {
w.WriteHeader(http.StatusNotFound)
return
}
b.gets++
w.Header().Set("Content-Length", strconv.Itoa(len(o.body)))
w.Write(o.body)
case http.MethodDelete:
delete(b.objects, key)
w.WriteHeader(http.StatusNoContent)
}
}))
t.Cleanup(srv.Close)
return srv, b
}
func (b *uvBucket) has(key string) (uvObject, bool) {
b.mu.Lock()
defer b.mu.Unlock()
o, ok := b.objects["files/"+key]
return o, ok
}
// uvEvents records what was raised, payload included.
type uvEvents struct {
mu sync.Mutex
raised []map[string]any
}
func (e *uvEvents) Raise(kind evt.Kind, repo string, payload map[string]any) error {
e.mu.Lock()
defer e.mu.Unlock()
if kind == evt.UVChanged {
p := map[string]any{"repo": repo}
for k, v := range payload {
p[k] = v
}
e.raised = append(e.raised, p)
}
return nil
}
type uvRig struct {
r *ops.Registry
db *oltp.DB
buckets *memBuckets
objects *uvBucket
events *uvEvents
url string
}
func newUVRig(t *testing.T, tweak ...func(*Deps)) uvRig {
t.Helper()
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
srv, objects := newUVBucket(t)
buckets := newMemBuckets()
events := &uvEvents{}
d := Deps{Version: "v", Repos: NewRepos(map[string]*Repo{"test": {}}), Buckets: buckets, UV: db, Events: events}
for _, f := range tweak {
f(&d)
}
r := ops.NewRegistry()
if err := RegisterAll(r, d); err != nil {
t.Fatal(err)
}
return uvRig{r: r, db: db, buckets: buckets, objects: objects, events: events, url: srv.URL}
}
func (g uvRig) bind(limit int64) {
g.buckets.rows["test"] = oltp.BucketBinding{Endpoint: g.url, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK", LimitBytes: limit}
}
func (g uvRig) put(ctx context.Context, name string, mtime int64, body string) (uvPutOut, error) {
out, err := g.r.Invoke(ctx, "uv.put", uvPutIn{Repo: "test", Name: name, Mtime: mtime,
Content: base64.StdEncoding.EncodeToString([]byte(body))})
if err != nil {
return uvPutOut{}, err
}
return out.(uvPutOut), nil
}
func (g uvRig) list(t *testing.T) uvListOut {
t.Helper()
out, err := g.r.Invoke(as("reader", "o"), "uv.list", uvListIn{Repo: "test"})
if err != nil {
t.Fatalf("FAIL: list: %v", err)
}
return out.(uvListOut)
}
// No bucket: the listing is empty and says so, and a write is refused
// in words that say what to do.
func TestUVNeedsABucket(t *testing.T) {
g := newUVRig(t)
l := g.list(t)
if l.Bound || len(l.Files) != 0 || l.Note == "" {
t.Errorf("FAIL: unbound list = %+v", l)
}
_, err := g.put(as("pub", "y"), "a.txt", 1, "x")
if !errors.Is(err, ops.ErrInvalid) || !strings.Contains(err.Error(), NoUVBucket) {
t.Errorf("FAIL: put with no bucket: %v", err)
}
if _, err := g.r.Invoke(as("pub", "y"), "uv.rm", uvRmIn{Repo: "test", Name: "a.txt"}); !errors.Is(err, ops.ErrInvalid) || !strings.Contains(err.Error(), NoUVBucket) {
t.Errorf("FAIL: rm with no bucket: %v", err)
}
// The forge's own bucket is not this repository's (RFC 0016), so it
// does not count as one.
g.buckets.rows[oltp.ForgeBucket] = oltp.BucketBinding{Endpoint: g.url, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK"}
if _, err := g.put(as("pub", "y"), "a.txt", 1, "x"); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: put into the forge's bucket: %v", err)
}
if len(g.objects.objects) != 0 {
t.Errorf("FAIL: objects written: %v", g.objects.objects)
}
t.Log("ok: no bucket, no unversioned files, and the refusal names the remedy")
}
// `y` writes; `a` alone does not, as in fossil; `s` does.
func TestUVPutNeedsY(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
for _, letters := range []string{"", "o", "a", "aoi"} {
if _, err := g.put(as("someone", letters), "a.txt", 1, "x"); !errors.Is(err, ops.ErrForbidden) {
t.Errorf("FAIL: put with %q: %v", letters, err)
}
if _, err := g.r.Invoke(as("someone", letters), "uv.rm", uvRmIn{Repo: "test", Name: "a.txt"}); !errors.Is(err, ops.ErrForbidden) {
t.Errorf("FAIL: rm with %q: %v", letters, err)
}
}
if _, err := g.put(as("root", "s"), "a.txt", 1, "x"); err != nil {
t.Errorf("FAIL: setup refused: %v", err)
}
if _, err := g.put(as("pub", "y"), "b.txt", 1, "x"); err != nil {
t.Errorf("FAIL: y refused: %v", err)
}
t.Log("ok: y and s publish, a does not")
}
func TestUVPutRefusesBadNamesAndOversize(t *testing.T) {
g := newUVRig(t)
g.bind(8)
pub := as("pub", "y")
for _, name := range []string{"has space", "../up", "/abs", "a\\b", "a//b", ".tombstones/x", strings.Repeat("n", 501)} {
if _, err := g.put(pub, name, 1, "x"); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: name %q: %v", name, err)
}
}
if _, err := g.put(pub, "big.bin", 1, "123456789"); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: over the limit: %v", err)
}
if _, err := g.put(pub, "fits.bin", 1, "12345678"); err != nil {
t.Errorf("FAIL: at the limit: %v", err)
}
bad := base64.StdEncoding.EncodeToString([]byte("x"))
if _, err := g.r.Invoke(pub, "uv.put", uvPutIn{Repo: "test", Name: "h.txt", Mtime: 1, Content: bad, Hash: strings.Repeat("0", 40)}); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: a hash that does not match: %v", err)
}
if len(g.objects.objects) != 1 {
t.Errorf("FAIL: refused writes reached the bucket: %d objects", len(g.objects.objects))
}
t.Log("ok: fossil's name rules, the binding's limit and the hash are checked before a byte is written")
}
// The whole round: put writes the object with its metadata and the
// row, list shows it, get presigns a download of unversioned/<repo>/<name>
// with the type from the table; HTML is never inline.
func TestUVPutListGet(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
out, err := g.put(as("pub", "y"), "releases/0.2.0/download.html", 1727740800, "<script>x</script>")
if err != nil {
t.Fatalf("FAIL: put: %v", err)
}
if out.File.Name != "releases/0.2.0/download.html" || out.File.Hash != uv.Hash([]byte("<script>x</script>")) || out.File.Size != 18 || out.File.Who != "pub" || out.File.Mtime != 1727740800 {
t.Errorf("FAIL: put = %+v", out)
}
o, ok := g.objects.has("unversioned/test/releases/0.2.0/download.html")
if !ok || string(o.body) != "<script>x</script>" || o.meta["mtime"] != "1727740800" || o.meta["hash"] != out.File.Hash || o.meta["who"] != "pub" {
t.Errorf("FAIL: object = %+v %v", o, ok)
}
if len(g.events.raised) != 1 || g.events.raised[0]["name"] != "releases/0.2.0/download.html" || g.events.raised[0]["deleted"] != false || g.events.raised[0]["repo"] != "test" {
t.Errorf("FAIL: events = %v", g.events.raised)
}
if _, err := g.put(as("pub", "y"), "notes.txt", 5, "hello"); err != nil {
t.Fatal(err)
}
l := g.list(t)
if !l.Bound || len(l.Files) != 2 || l.Files[0].Name != "notes.txt" || l.Files[1].Name != "releases/0.2.0/download.html" {
t.Errorf("FAIL: list = %+v", l)
}
if l.Hash != uv.CatalogHash(l.Files) {
t.Errorf("FAIL: list hash %s is not the catalog's", l.Hash)
}
got, err := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "releases/0.2.0/download.html"})
if err != nil {
t.Fatalf("FAIL: get: %v", err)
}
dl := got.(uvGetOut)
u, err := url.Parse(dl.URL)
if err != nil || u.Path != "/files/unversioned/test/releases/0.2.0/download.html" || u.Query().Get("X-Amz-Signature") == "" {
t.Errorf("FAIL: url = %s", dl.URL)
}
if !strings.HasPrefix(u.Query().Get("response-content-disposition"), "attachment;") || u.Query().Get("response-content-type") != "text/html; charset=utf-8" || dl.Inline {
t.Errorf("FAIL: html is not forced to download: %v inline=%v", u.Query(), dl.Inline)
}
got, err = g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "notes.txt"})
if err != nil {
t.Fatal(err)
}
if txt := got.(uvGetOut); !txt.Inline || !strings.Contains(txt.URL, "response-content-disposition=inline") {
t.Errorf("FAIL: text is not shown in place: %+v", txt)
}
if _, err := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "nothing"}); !errors.Is(err, ops.ErrNotFound) {
t.Errorf("FAIL: get of an absent name: %v", err)
}
t.Log("ok: put, list and get agree, and the download is shaped by the table")
}
// Newest mtime wins; a removal is a tombstone row and object; a newer
// put revives the name.
func TestUVMtimeRuleAndTombstones(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
pub := as("pub", "y")
if _, err := g.put(pub, "a.txt", 100, "new"); err != nil {
t.Fatal(err)
}
if _, err := g.put(pub, "a.txt", 99, "old"); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: an older put: %v", err)
}
if o, _ := g.objects.has("unversioned/test/a.txt"); string(o.body) != "new" {
t.Errorf("FAIL: the older put reached the bucket: %q", o.body)
}
if _, err := g.r.Invoke(pub, "uv.rm", uvRmIn{Repo: "test", Name: "never-was"}); !errors.Is(err, ops.ErrNotFound) {
t.Errorf("FAIL: rm of an absent name: %v", err)
}
if _, err := g.r.Invoke(as("dan", "y"), "uv.rm", uvRmIn{Repo: "test", Name: "a.txt", Mtime: 200}); err != nil {
t.Fatalf("FAIL: rm: %v", err)
}
l := g.list(t)
if len(l.Files) != 1 || !l.Files[0].Deleted || l.Files[0].Mtime != 200 || l.Files[0].Who != "dan" || l.Files[0].Hash != "" {
t.Errorf("FAIL: after rm = %+v", l.Files)
}
if tomb, ok := g.objects.has("unversioned/test/.tombstones/a.txt"); !ok || len(tomb.body) != 0 || tomb.meta["mtime"] != "200" {
t.Errorf("FAIL: tombstone object = %+v %v", tomb, ok)
}
if _, ok := g.objects.has("unversioned/test/a.txt"); ok {
t.Error("FAIL: the removed content is still in the bucket")
}
if _, err := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "a.txt"}); !errors.Is(err, ops.ErrNotFound) {
t.Errorf("FAIL: get of a tombstone: %v", err)
}
if l.Hash != uv.CatalogHash(nil) {
t.Errorf("FAIL: a tombstone counted in the uv-hash")
}
if _, err := g.put(pub, "a.txt", 150, "between"); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: a put older than the tombstone: %v", err)
}
if _, err := g.put(pub, "a.txt", 300, "back"); err != nil {
t.Fatalf("FAIL: revive: %v", err)
}
l = g.list(t)
if len(l.Files) != 1 || l.Files[0].Deleted || l.Files[0].Mtime != 300 {
t.Errorf("FAIL: after revive = %+v", l.Files)
}
if _, ok := g.objects.has("unversioned/test/.tombstones/a.txt"); ok {
t.Error("FAIL: the tombstone outlived the revival")
}
if n := len(g.events.raised); n != 3 || g.events.raised[1]["deleted"] != true {
t.Errorf("FAIL: events = %v", g.events.raised)
}
t.Log("ok: newest mtime wins, removals are tombstones, and a newer put revives")
}
// The catalog is disposable: dropped, it comes back from the bucket's
// listing and metadata identical, tombstones and authors included; a
// different bucket bound rebuilds it from that bucket.
func TestUVCatalogRebuildsFromTheBucket(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
pub := as("pub", "y")
for i, name := range []string{"a.txt", "releases/1/x.tgz", "releases/1/SHA256SUMS", "z.md"} {
if _, err := g.put(pub, name, int64(100+i), "body of "+name); err != nil {
t.Fatal(err)
}
}
if _, err := g.r.Invoke(as("dan", "y"), "uv.rm", uvRmIn{Repo: "test", Name: "z.md", Mtime: 500}); err != nil {
t.Fatal(err)
}
before := g.list(t)
if err := g.db.DropUVCatalog("test"); err != nil {
t.Fatal(err)
}
listsBefore := g.objects.lists
after := g.list(t)
if g.objects.lists <= listsBefore {
t.Fatal("FAIL: the listing did not consult the bucket")
}
if len(after.Files) != len(before.Files) || after.Hash != before.Hash {
t.Fatalf("FAIL: rebuilt %+v, was %+v", after.Files, before.Files)
}
for i := range before.Files {
b, a := before.Files[i], after.Files[i]
if a.Name != b.Name || a.Mtime != b.Mtime || a.Hash != b.Hash || a.Size != b.Size || a.Deleted != b.Deleted || a.Who != b.Who {
t.Errorf("FAIL: %s rebuilt as %+v, was %+v", b.Name, a, b)
}
}
// A second read does not list again: the catalog knows its source.
lists := g.objects.lists
g.list(t)
if g.objects.lists != lists {
t.Error("FAIL: a built catalog listed the bucket on a read")
}
// The explicit hook does the same, for an admin.
if _, err := g.r.Invoke(as("pub", "y"), "uv.rebuild", uvRebuildIn{Repo: "test"}); !errors.Is(err, ops.ErrForbidden) {
t.Errorf("FAIL: rebuild without a: %v", err)
}
rb, err := g.r.Invoke(as("admin", "a"), "uv.rebuild", uvRebuildIn{Repo: "test"})
if err != nil || rb.(uvListOut).Hash != before.Hash || len(rb.(uvListOut).Files) != 4 {
t.Errorf("FAIL: rebuild = %+v %v", rb, err)
}
// Another bucket bound: the catalog is that bucket's, here empty.
other := g.buckets.rows["test"]
other.Bucket = "elsewhere"
g.buckets.rows["test"] = other
if l := g.list(t); len(l.Files) != 0 {
t.Errorf("FAIL: a new bucket kept the old catalog: %+v", l.Files)
}
t.Log("ok: drop the catalog and it comes back from the bucket, tombstones included")
}
// A hash a fossil client sends under the SHA1 policy is kept as sent,
// so the uv-hash agrees with that client's.
func TestUVPutKeepsASHA1Hash(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
sum := sha1.Sum([]byte("old policy"))
h := hex.EncodeToString(sum[:])
out, err := g.r.Invoke(as("pub", "y"), "uv.put", uvPutIn{Repo: "test", Name: "a.txt", Mtime: 7,
Content: base64.StdEncoding.EncodeToString([]byte("old policy")), Hash: h})
if err != nil || out.(uvPutOut).File.Hash != h {
t.Fatalf("FAIL: put = %+v %v", out, err)
}
t.Log("ok: a SHA1 hash is verified and kept")
}
// A tracking mirror takes no local writes, unversioned ones included.
func TestUVPutRefusedOnATrackingMirror(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
d := Deps{Version: "v", Repos: NewRepos(map[string]*Repo{"test": {}}), Buckets: g.buckets, UV: g.db,
Mirrors: memMirrors{"test": {Repo: "test", URL: "https://up.example/x", Enabled: true}}}
r := ops.NewRegistry()
if err := RegisterAll(r, d); err != nil {
t.Fatal(err)
}
_, err := r.Invoke(as("pub", "y"), "uv.put", uvPutIn{Repo: "test", Name: "a.txt", Mtime: 1, Content: "eA=="})
if !errors.Is(err, ops.ErrForbidden) {
t.Errorf("FAIL: put on a tracking mirror: %v", err)
}
t.Log("ok: a tracking mirror refuses unversioned writes")
}
// countingBody reports how much of a request body was read.
type countingBody struct {
r io.Reader
read int
}
func (c *countingBody) Read(p []byte) (int, error) {
n, err := c.r.Read(p)
c.read += n
return n, err
}
func (c *countingBody) Close() error { return nil }
// uv.put's body is bounded by the binding's limit before it is read:
// a body far past it is refused with 413 having read no more than the
// bound, whether or not it declares its length, and one inside it goes
// through.
func TestUVPutBodyIsBoundedBeforeItIsRead(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 10)
h, err := Router(g.r, "v")
if err != nil {
t.Fatal(err)
}
huge := `{"name":"big.bin","mtime":1,"content":"` + strings.Repeat("A", 8<<20) + `"}`
for _, declared := range []bool{true, false} {
body := &countingBody{r: strings.NewReader(huge)}
req := httptest.NewRequest(http.MethodPost, "/api/v1/r/test/uv", nil)
req.Body = body
req.ContentLength = -1
if declared {
req.ContentLength = int64(len(huge))
}
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusRequestEntityTooLarge {
t.Errorf("FAIL: declared=%v: status %d %s", declared, rec.Code, rec.Body)
}
if int64(body.read) > uvBodyLimit(1<<10)+64<<10 {
t.Errorf("FAIL: declared=%v: read %d bytes of an oversized body", declared, body.read)
}
}
small := `{"name":"ok.bin","mtime":1,"content":"` + base64.StdEncoding.EncodeToString([]byte("fits")) + `"}`
req := httptest.NewRequest(http.MethodPost, "/api/v1/r/test/uv", strings.NewReader(small))
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Errorf("FAIL: a small put: %d %s", rec.Code, rec.Body)
}
t.Log("ok: an oversized upload is refused before it is read whole")
}
// What the bucket stores is what uv.get would answer: anything not
// shown in place is stored as bytes with an attachment disposition, so
// a public origin, which signs nothing, serves it as a download too.
func TestUVStoresActiveTypesAsDownloads(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
pub := as("pub", "y")
for _, name := range []string{"download.html", "logo.svg", "app.js", "feed.xml", "doc.pdf", "site.css", "SHA256SUMS", "notes.txt", "shot.png"} {
if _, err := g.put(pub, name, 1, "x"); err != nil {
t.Fatal(err)
}
o, _ := g.objects.has("unversioned/test/" + name)
_, inline := uv.ContentType(name)
switch {
case inline && (o.disposition != "" || o.ctype == "application/octet-stream"):
t.Errorf("FAIL: %s is passive and stored as %q %q", name, o.ctype, o.disposition)
case !inline && (o.ctype != "application/octet-stream" || !strings.HasPrefix(o.disposition, "attachment;")):
t.Errorf("FAIL: %s is stored as %q %q", name, o.ctype, o.disposition)
}
}
t.Log("ok: the stored object is a download wherever uv.get says download")
}
// A private repository's files must not sit behind a public origin,
// where a guessable key is a public URL: writes are refused in words
// that say why, and a read never links the public base.
func TestUVPrivateRepoRefusesAPublicBucket(t *testing.T) {
private := false
g := newUVRig(t, func(d *Deps) {
d.Private = func() map[string]bool { return map[string]bool{"test": private} }
})
g.bind(1 << 20)
b := g.buckets.rows["test"]
b.PublicBase = "https://files.example"
g.buckets.rows["test"] = b
pub := as("pub", "y")
if _, err := g.put(pub, "notes.txt", 1, "x"); err != nil {
t.Fatalf("FAIL: a public repository refused: %v", err)
}
got, _ := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "notes.txt"})
if !strings.HasPrefix(got.(uvGetOut).URL, "https://files.example/unversioned/test/notes.txt") {
t.Errorf("FAIL: a public repository's passive file did not use the public base: %s", got.(uvGetOut).URL)
}
private = true
if _, err := g.put(pub, "more.txt", 2, "x"); !errors.Is(err, ops.ErrInvalid) || !strings.Contains(err.Error(), "public") {
t.Errorf("FAIL: put on a private repository with a public base: %v", err)
}
if _, err := g.r.Invoke(pub, "uv.rm", uvRmIn{Repo: "test", Name: "notes.txt", Mtime: 5}); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: rm on a private repository with a public base: %v", err)
}
if _, ok := g.objects.has("unversioned/test/more.txt"); ok {
t.Error("FAIL: the refused file reached the bucket")
}
got, err := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: "test", Name: "notes.txt"})
if err != nil || strings.HasPrefix(got.(uvGetOut).URL, "https://files.example") || got.(uvGetOut).Expires == "" {
t.Errorf("FAIL: a private repository was linked on its public base: %+v %v", got, err)
}
t.Log("ok: a private repository publishes nothing behind a public origin")
}
// failingCatalog is the real catalog with its next write failing, as a
// crash or a full disk between the bucket write and the row would.
type failingCatalog struct {
*oltp.DB
fail bool
}
func (f *failingCatalog) PutUVFile(repo string, file uv.File) (bool, error) {
if f.fail {
f.fail = false
return false, errors.New("disk I/O error")
}
return f.DB.PutUVFile(repo, file)
}
// The bucket write succeeded and the row did not: the catalog no longer
// describes the bucket, so it is dropped and the next read rebuilds it
// from the bucket, which has the new bytes.
func TestUVCatalogFailureAfterTheBucketWriteRebuilds(t *testing.T) {
cat := &failingCatalog{}
g := newUVRig(t, func(d *Deps) {
cat.DB = d.UV.(*oltp.DB)
d.UV = cat
})
g.bind(1 << 20)
pub := as("pub", "y")
if _, err := g.put(pub, "a.txt", 1, "first"); err != nil {
t.Fatal(err)
}
cat.fail = true
if _, err := g.put(pub, "a.txt", 2, "second"); err == nil {
t.Fatal("FAIL: a failed catalog write reported success")
}
if _, built, _ := g.db.UVSource("test"); built {
t.Error("FAIL: the catalog was kept after it stopped describing the bucket")
}
l := g.list(t)
if len(l.Files) != 1 || l.Files[0].Mtime != 2 || l.Files[0].Hash != uv.Hash([]byte("second")) {
t.Errorf("FAIL: rebuilt = %+v", l.Files)
}
t.Log("ok: a catalog that missed a write is rebuilt from the bucket")
}
// A name a listing has to URL-encode survives a rebuild, whether the
// service encodes keys or sends them as they are; and the object
// carries its name, so the rebuild does not depend on either.
func TestUVRebuildKeepsEncodedNames(t *testing.T) {
for _, raw := range []bool{false, true} {
t.Run(fmt.Sprintf("raw=%v", raw), func(t *testing.T) { rebuildKeepsNames(t, raw) })
}
}
func rebuildKeepsNames(t *testing.T, raw bool) {
g := newUVRig(t)
g.objects.raw = raw
g.bind(1 << 20)
names := []string{"100%.txt", "a+b.txt", "c++.tgz", "ü/é.txt"}
for _, name := range names {
if _, err := g.put(as("pub", "y"), name, 1, name); err != nil {
t.Fatal(err)
}
}
if _, err := g.r.Invoke(as("pub", "y"), "uv.rm", uvRmIn{Repo: "test", Name: "a+b.txt", Mtime: 9}); err != nil {
t.Fatal(err)
}
if o, _ := g.objects.has("unversioned/test/c++.tgz"); o.meta["uv-name"] != "c%2B%2B.tgz" {
t.Errorf("FAIL: the object does not carry its name: %v", o.meta)
}
before := g.list(t)
if err := g.db.DropUVCatalog("test"); err != nil {
t.Fatal(err)
}
after := g.list(t)
if after.Hash != before.Hash || len(after.Files) != len(names) {
t.Errorf("FAIL: rebuilt %+v, was %+v", after.Files, before.Files)
}
for i := range before.Files {
if i < len(after.Files) && (after.Files[i].Name != before.Files[i].Name || after.Files[i].Deleted != before.Files[i].Deleted) {
t.Errorf("FAIL: %+v rebuilt as %+v", before.Files[i], after.Files[i])
}
}
t.Log("ok: encoded keys come back as the names they were")
}
// The body bound depends on the repository's binding, so it must not
// answer before the gates do: a stranger sending the same oversized
// put to a private repository with a bucket and to a name that does not
// exist gets the same answer, anonymously (401) and signed in without
// read (404, the repository named as the caller named it).
func TestUVPutBodyBoundIsNoExistenceOracle(t *testing.T) {
g := newUVRig(t, func(d *Deps) {
d.Private = func() map[string]bool { return map[string]bool{"test": true} }
})
g.bind(oltp.DefaultBucketLimit)
verify := func(repo, user, pass string, need ops.Auth) Decision {
if user == "stranger" && pass == "pw" {
return Allowed
}
return Denied
}
capsFor := func(account, repo string) CapSet { return oltp.CapSet("") }
canRead := func(ctx context.Context, repo string) bool { return false }
h, err := RouterWithCaps(g.r, "v", capsFor, canRead, verify)
if err != nil {
t.Fatal(err)
}
body := `{"name":"big.bin","mtime":1,"content":"` + strings.Repeat("A", 1<<20) + `"}`
ask := func(repo string, signedIn bool) (int, string) {
req := httptest.NewRequest(http.MethodPost, "/api/v1/r/"+repo+"/uv", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
if signedIn {
req.SetBasicAuth("stranger", "pw")
}
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
return rec.Code, strings.ReplaceAll(rec.Body.String(), repo, "<repo>")
}
for _, signedIn := range []bool{false, true} {
realCode, realBody := ask("test", signedIn)
ghostCode, ghostBody := ask("ghost", signedIn)
if realCode != ghostCode || realBody != ghostBody {
t.Errorf("FAIL: signedIn=%v: private repo answered %d %q, missing one %d %q", signedIn, realCode, realBody, ghostCode, ghostBody)
}
if realCode == http.StatusRequestEntityTooLarge {
t.Errorf("FAIL: signedIn=%v: the body bound answered before the gates", signedIn)
}
}
t.Log("ok: an oversized put tells a stranger nothing about which repositories exist")
}
// The name an object carries wins over its key as a listing returned
// it, so a service that decoded a "+" into a space cannot rename a
// file; an object written without one falls back to its key.
func TestUVNameOfPrefersTheObjectsOwnName(t *testing.T) {
for _, c := range []struct {
key, meta, name string
tombstone, ok bool
}{
{"unversioned/q/c .tgz", "c%2B%2B.tgz", "c++.tgz", false, true},
{"unversioned/q/.tombstones/c .tgz", "c%2B%2B.tgz", "c++.tgz", true, true},
{"unversioned/q/plain.txt", "", "plain.txt", false, true},
{"unversioned/q/bad name", "", "", false, false},
{"unversioned/q/x.txt", "has%20space", "x.txt", false, true},
} {
name, tomb, ok := uvNameOf("q", c.key, c.meta)
if name != c.name || tomb != c.tombstone || ok != c.ok {
t.Errorf("FAIL: uvNameOf(%q, %q) = %q %v %v", c.key, c.meta, name, tomb, ok)
}
}
t.Log("ok: the object's own name wins")
}
// A removal can name the version it saw: refused if the file has been
// replaced since, and dated after that version so a file published this
// second can be removed this second.
func TestUVRmIfMtime(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
pub := as("pub", "y")
now := time.Now().Unix() + 3600
if _, err := g.put(pub, "a.txt", now, "a"); err != nil {
t.Fatal(err)
}
if _, err := g.r.Invoke(pub, "uv.rm", uvRmIn{Repo: "test", Name: "a.txt", IfMtime: now - 1}); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: a stale precondition = %v", err)
}
out, err := g.r.Invoke(pub, "uv.rm", uvRmIn{Repo: "test", Name: "a.txt", IfMtime: now})
if err != nil || out.(uvPutOut).File.Mtime != now+1 {
t.Errorf("FAIL: rm with the seen version = %+v %v", out, err)
}
t.Log("ok: the precondition holds and the tombstone follows the version")
}
func uvForm(t *testing.T, fields [][2]string, filename, content string) *http.Request {
t.Helper()
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
fw, _ := mw.CreateFormFile("file", filename)
fw.Write([]byte(content))
for _, f := range fields {
mw.WriteField(f[0], f[1])
}
mw.Close()
req := httptest.NewRequest("POST", "/r/test/uv", &buf)
req.Header.Set("Content-Type", mw.FormDataContentType())
return req
}
// The form path streams the file into uv.put with the op's gates, its
// fields read after the file as a browser sends them, bounded by the
// binding's limit before uv.put is asked.
func TestUVUploadForm(t *testing.T) {
g := newUVRig(t)
g.bind(16)
d := Deps{Repos: NewRepos(map[string]*Repo{"test": {}}), Buckets: g.buckets, UV: g.db}
name, err := UVUploadForm(as("pub", "y"), d, g.r, "test", httptest.NewRecorder(), uvForm(t, [][2]string{{"prefix", "releases/1.0/"}}, `C:\dl\z.bin`, "zz"))
if err != nil || name != "releases/1.0/z.bin" {
t.Fatalf("FAIL: upload = %q %v", name, err)
}
if o, ok := g.objects.has("unversioned/test/releases/1.0/z.bin"); !ok || string(o.body) != "zz" {
t.Errorf("FAIL: the object = %+v", o)
}
if name, err := UVUploadForm(as("pub", "y"), d, g.r, "test", httptest.NewRecorder(), uvForm(t, [][2]string{{"name", "n.bin"}}, "x", "n")); err != nil || name != "n.bin" {
t.Errorf("FAIL: named upload = %q %v", name, err)
}
if _, err := UVUploadForm(as("pub", "y"), d, g.r, "test", httptest.NewRecorder(), uvForm(t, nil, "big.bin", strings.Repeat("b", 17))); !errors.Is(err, ops.ErrInvalid) {
t.Errorf("FAIL: over the limit = %v", err)
}
if _, err := UVUploadForm(as("pub", "a"), d, g.r, "test", httptest.NewRecorder(), uvForm(t, nil, "a.bin", "a")); !errors.Is(err, ops.ErrForbidden) {
t.Errorf("FAIL: without y = %v", err)
}
t.Log("ok: the form streams into uv.put, bounded, gated by the op")
}
// The form path bounds the whole body, not only the file part: a
// request past the limit is cut off however its parts are named.
func TestUVUploadFormBoundsTheBody(t *testing.T) {
g := newUVRig(t)
g.bind(16)
d := Deps{Repos: NewRepos(map[string]*Repo{"test": {}}), Buckets: g.buckets, UV: g.db}
big := strings.Repeat("p", 200<<10)
req := uvForm(t, [][2]string{{"padding", big}}, "a.bin", "a")
if _, err := UVUploadForm(as("pub", "y"), d, g.r, "test", httptest.NewRecorder(), req); err == nil {
t.Errorf("FAIL: a body past the limit was read whole")
}
if _, err := UVUploadForm(as("pub", "y"), d, g.r, "test", httptest.NewRecorder(), uvForm(t, nil, "a.bin", "a")); err != nil {
t.Errorf("FAIL: a small upload = %v", err)
}
t.Log("ok: the whole body is bounded")
}
// Several repositories may bind one bucket (RFC 0016), so each one's
// unversioned files live under its own prefix: the same name in two of
// them is two files, and a rebuild reads only its own.
func TestUVRepositoriesShareABucket(t *testing.T) {
g := newUVRig(t, func(d *Deps) { d.Repos = NewRepos(map[string]*Repo{"test": {}, "other": {}}) })
g.bind(1 << 20)
g.buckets.rows["other"] = g.buckets.rows["test"]
pub := as("pub", "y")
put := func(repo, name, body string) {
t.Helper()
if _, err := g.r.Invoke(pub, "uv.put", uvPutIn{Repo: repo, Name: name, Mtime: 100,
Content: base64.StdEncoding.EncodeToString([]byte(body))}); err != nil {
t.Fatalf("FAIL: put %s %s: %v", repo, name, err)
}
}
list := func(repo string) []uv.File {
t.Helper()
out, err := g.r.Invoke(as("reader", "o"), "uv.list", uvListIn{Repo: repo})
if err != nil {
t.Fatalf("FAIL: list %s: %v", repo, err)
}
return out.(uvListOut).Files
}
put("test", "same.txt", "test's")
put("other", "same.txt", "other's")
put("other", "only-other.txt", "x")
for repo, body := range map[string]string{"test": "test's", "other": "other's"} {
if o, ok := g.objects.has("unversioned/" + repo + "/same.txt"); !ok || string(o.body) != body {
t.Errorf("FAIL: %s's same.txt in the bucket = %q %v", repo, o.body, ok)
}
got, err := g.r.Invoke(as("reader", "o"), "uv.get", uvGetIn{Repo: repo, Name: "same.txt"})
if err != nil {
t.Fatal(err)
}
u, err := url.Parse(got.(uvGetOut).URL)
if err != nil || u.Path != "/files/unversioned/"+repo+"/same.txt" || got.(uvGetOut).File.Hash != uv.Hash([]byte(body)) {
t.Errorf("FAIL: %s's download = %+v", repo, got)
}
}
if l := list("test"); len(l) != 1 || l[0].Name != "same.txt" || l[0].Hash != uv.Hash([]byte("test's")) {
t.Errorf("FAIL: test lists %+v", l)
}
if l := list("other"); len(l) != 2 {
t.Errorf("FAIL: other lists %+v", l)
}
if _, err := g.r.Invoke(as("dan", "y"), "uv.rm", uvRmIn{Repo: "test", Name: "same.txt", Mtime: 200}); err != nil {
t.Fatal(err)
}
if _, ok := g.objects.has("unversioned/test/.tombstones/same.txt"); !ok {
t.Error("FAIL: test's tombstone is not under its prefix")
}
if o, ok := g.objects.has("unversioned/other/same.txt"); !ok || string(o.body) != "other's" {
t.Errorf("FAIL: test's removal touched other's file: %q %v", o.body, ok)
}
// An object outside a repository's prefix, as the unscoped layout
// wrote them, is no repository's.
g.objects.mu.Lock()
g.objects.objects["files/unversioned/stray.txt"] = uvObject{body: []byte("s"),
meta: map[string]string{uvMetaMtime: "100", uvMetaHash: uv.Hash([]byte("s"))}}
g.objects.mu.Unlock()
for _, repo := range []string{"test", "other"} {
if err := g.db.DropUVCatalog(repo); err != nil {
t.Fatal(err)
}
}
if l := list("test"); len(l) != 1 || l[0].Name != "same.txt" || !l[0].Deleted {
t.Errorf("FAIL: test rebuilt as %+v", l)
}
l := list("other")
if len(l) != 2 || l[0].Name != "only-other.txt" || l[1].Name != "same.txt" || l[1].Deleted || l[1].Hash != uv.Hash([]byte("other's")) {
t.Errorf("FAIL: other rebuilt as %+v", l)
}
t.Log("ok: two repositories on one bucket keep, remove and rebuild only their own files")
}
// A put of exactly what the catalog holds (same name, mtime and hash) is
// nothing to do, as in fossil: answered as success with applied false,
// no write and no event. An older mtime, or the same mtime with a hash
// that loses, is still refused in words.
func TestUVPutOfTheSameVersionIsIdempotent(t *testing.T) {
g := newUVRig(t)
g.bind(1 << 20)
pub := as("pub", "y")
first, err := g.put(pub, "a.txt", 100, "same")
if err != nil || !first.Applied {
t.Fatalf("FAIL: first put = %+v %v", first, err)
}
again, err := g.put(as("dan", "y"), "a.txt", 100, "same")
if err != nil || again.Applied || again.File.Hash != first.File.Hash || again.File.Who != "pub" {
t.Errorf("FAIL: the same put again = %+v %v; want success, applied false, the held version", again, err)
}
if o, _ := g.objects.has("unversioned/test/a.txt"); o.meta["who"] != "pub" {
t.Errorf("FAIL: the repeat rewrote the object: %v", o.meta)
}
if n := len(g.events.raised); n != 1 {
t.Errorf("FAIL: events = %d, want 1", n)
}
if _, err := g.put(pub, "a.txt", 99, "same"); !errors.Is(err, ops.ErrInvalid) || !strings.Contains(err.Error(), "already holds") {
t.Errorf("FAIL: the same content at an older mtime: %v", err)
}
loser, winner := "b", "c"
if uv.Hash([]byte(loser)) > uv.Hash([]byte(winner)) {
loser, winner = winner, loser
}
if _, err := g.put(pub, "b.txt", 100, winner); err != nil {
t.Fatal(err)
}
if _, err := g.put(pub, "b.txt", 100, loser); !errors.Is(err, ops.ErrInvalid) || !strings.Contains(err.Error(), "already holds") {
t.Errorf("FAIL: the same mtime with a losing hash: %v", err)
}
t.Log("ok: a repeated put is nothing to do; an older or losing one is refused")
}