package xfer
import (
"bytes"
"compress/zlib"
"context"
"crypto/sha1"
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os/exec"
"strings"
"testing"
"craftdesign.group/quarry/internal/uv"
)
// fakeUV is a catalog and its bytes, with y held by the named users.
type fakeUV struct {
files []uv.File
content map[string][]byte
y map[string]bool
wrote []uv.File
// failOn makes the write of that name fail as a bucket would.
failOn string
// tooLarge makes reading these names fail as over the limit.
tooLarge map[string]bool
}
func (u *fakeUV) UVCatalog(context.Context, string) ([]uv.File, bool, error) {
return u.files, true, nil
}
func (u *fakeUV) UVWritable(context.Context, string) (int64, error) { return 8, nil }
func (u *fakeUV) UVContent(_ context.Context, _ string, f uv.File) ([]byte, error) {
if u.tooLarge[f.Name] {
return nil, fmt.Errorf("%s: %w", f.Name, ErrUVTooLarge)
}
return u.content[f.Name], nil
}
func (u *fakeUV) UVWrite(_ context.Context, _ string, f uv.File, _ []byte) (uv.File, bool, error) {
if f.Name == u.failOn {
return uv.File{}, false, errors.New("bucket unavailable")
}
u.wrote = append(u.wrote, f)
return f, true, nil
}
func (u *fakeUV) UVTouch(_ context.Context, _ string, f uv.File) (uv.File, bool, error) {
u.wrote = append(u.wrote, f)
return f, true, nil
}
func (u *fakeUV) MayPublish(_, user string) bool { return u.y[user] }
func newFakeUV() *fakeUV {
body := []byte("hello\n")
return &fakeUV{
files: []uv.File{
{Name: "gone.txt", Mtime: 5, Deleted: true},
{Name: "hello.txt", Mtime: 7, Hash: uv.Hash(body), Size: int64(len(body))},
},
content: map[string][]byte{"hello.txt": body},
y: map[string]bool{"pub": true},
}
}
// The catalog goes out as fossil's send_unversioned_catalog sends it, a
// tombstone as hash "-" and size 0; a uvgimme for a tombstone is the
// deletion card; and once the reply reaches fossil's budget a uvgimme
// is answered with uvigot, so the client asks again next round.
func TestUVAnswerFollowsFossilsServer(t *testing.T) {
u := newFakeUV()
f := New()
f.SetUnversioned(u)
hello := u.files[1]
pr := scanRequest([]byte("pragma uv-hash " + strings.Repeat("0", 40) + "\nuvgimme gone.txt\nuvgimme hello.txt\nuvgimme absent.txt\n"))
out, err := f.uvAnswer(context.Background(), "r", pr, uvWho{}, 0)
if err != nil {
t.Fatal(err)
}
want := "pragma uv-pull-only\n" +
"uvigot gone.txt 5 - 0\n" +
"uvigot hello.txt 7 " + hello.Hash + " 6\n" +
"uvfile gone.txt 5 - 0 1\n" +
"uvfile hello.txt 7 " + hello.Hash + " 6 0\nhello\n"
if string(out) != want {
t.Errorf("FAIL: reply =\n%q\nwant\n%q", out, want)
} else {
t.Log("ok: catalog, tombstone and content are fossil's cards")
}
out, _ = f.uvAnswer(context.Background(), "r", pr, uvWho{user: "pub"}, uvReplyBudget)
if !strings.HasPrefix(string(out), "pragma uv-push-ok\n") || !strings.HasSuffix(string(out), "uvigot hello.txt 7 "+hello.Hash+" 6\n") {
t.Errorf("FAIL: over budget =\n%s", out)
} else {
t.Log("ok: y draws uv-push-ok, and a full reply defers the content to the next round")
}
pr = scanRequest([]byte("pragma uv-hash " + uv.CatalogHash(u.files) + "\n"))
if out, _ := f.uvAnswer(context.Background(), "r", pr, uvWho{user: "pub"}, 0); len(out) != 0 {
t.Errorf("FAIL: a matching uv-hash drew %q", out)
} else {
t.Log("ok: a matching uv-hash draws nothing")
}
}
// Refusals come before anything is applied, in fossil's words with the
// letter named; each kind of uvfile is applied as fossil applies it.
func TestUVRefusalAndApply(t *testing.T) {
u := newFakeUV()
f := New()
f.SetUnversioned(u)
ctx := context.Background()
good := "uvfile a.txt 9 " + uv.Hash([]byte("abc")) + " 3 0\nabc"
for name, c := range map[string]struct{ login, body, says string }{
"anonymous": {"", good, "Write permissions for unversioned files missing"},
"without y": {"dev", good, "y (Write-UV)"},
"over the limit": {"pub", "uvfile big.txt 9 " + uv.Hash([]byte("123456789")) + " 9 0\n123456789", "over this bucket's limit of 8"},
"a wrong hash": {"pub", "uvfile a.txt 9 " + uv.Hash([]byte("xyz")) + " 3 0\nabc", "HASH does not match CONTENT"},
"a bad name": {"pub", "uvfile ../a 9 - 0 1\n", "malformed uvfile line"},
"a bad hash": {"pub", "uvfile a.txt 9 nothex 3 4\n", "malformed uvfile line"},
} {
why := f.uvRefusal(ctx, "r", scanRequest([]byte(c.body)), uvWho{user: c.login})
if !strings.Contains(why, c.says) {
t.Errorf("FAIL: %s: refusal %q, want %q", name, why, c.says)
} else {
t.Logf("ok: %s is refused: %s", name, why)
}
}
if len(u.wrote) != 0 {
t.Fatalf("FAIL: a refusal wrote %v", u.wrote)
}
f.SetWriteGate(func(string) (bool, string) { return false, "https://up.example" })
if why := f.uvRefusal(ctx, "r", scanRequest([]byte(good)), uvWho{user: "pub"}); !strings.Contains(why, "mirrors https://up.example") {
t.Errorf("FAIL: tracking mirror: %q", why)
} else {
t.Log("ok: a tracking mirror refuses a uvfile as a write")
}
f.SetWriteGate(nil)
body := good + "uvfile b.txt 10 " + uv.Hash([]byte("hello\n")) + " 6 4\nuvfile c.txt 11 - 0 1\n"
pr := scanRequest([]byte(body))
if why := f.uvRefusal(ctx, "r", pr, uvWho{user: "pub"}); why != "" {
t.Fatalf("FAIL: refused %q", why)
}
if _, err := f.uvAnswer(ctx, "r", pr, uvWho{user: "pub"}, 0); err != nil {
t.Fatal(err)
}
w := u.wrote
if len(w) != 3 || w[0].Name != "a.txt" || w[0].Who != "pub" || w[1].Name != "b.txt" || w[1].Mtime != 10 || !w[2].Deleted || w[2].Mtime != 11 {
t.Errorf("FAIL: applied %+v", w)
} else {
t.Log("ok: content, mtime-only and tombstone are applied in order, as the login")
}
}
// signed is body with a login card for user inserted at the given
// offset, signed over what follows it the way a client signs.
func signed(t *testing.T, before, after, user, secret string) []byte {
t.Helper()
nonce := sha1Hex([]byte(after))
return []byte(before + "login " + user + " " + nonce + " " + sha1Hex([]byte(nonce+secret)) + "\n" + after)
}
func sha1Hex(b []byte) string {
h := sha1.Sum(b)
return hex.EncodeToString(h[:])
}
// A card before the login card is not covered by its signature, so it
// speaks for nobody, as it would in fossil's in-order reading: a uvfile
// there is refused even when the login holds y, and a uv-hash there is
// answered as nobody.
func TestUVCardsBeforeTheLoginAreNobodys(t *testing.T) {
u := newFakeUV()
f := New()
f.SetUnversioned(u)
ctx := context.Background()
write := "uvfile a.txt 9 " + uv.Hash([]byte("abc")) + " 3 0\nabc"
body := signed(t, write, "pragma client-version 1\n", "pub", "s")
pr := scanRequest(body)
who := uvWho{user: "pub", from: pr.coveredFrom}
if why := f.uvRefusal(ctx, "r", pr, who); !strings.Contains(why, "before the login card") {
t.Errorf("FAIL: a uvfile before the login: %q", why)
} else {
t.Log("ok: a uvfile before the login card is refused though the login holds y")
}
pr = scanRequest(signed(t, "", write, "pub", "s"))
if why := f.uvRefusal(ctx, "r", pr, uvWho{user: "pub", from: pr.coveredFrom}); why != "" {
t.Errorf("FAIL: the same uvfile after the login was refused: %q", why)
} else {
t.Log("ok: the same uvfile after the login card is the login's")
}
cookie := uvWho{user: "pub", from: 0} // the cookie form signs the whole body
if why := f.uvRefusal(ctx, "r", scanRequest([]byte(write)), cookie); why != "" {
t.Errorf("FAIL: a cookie login does not cover the body: %q", why)
}
hash := "pragma uv-hash " + strings.Repeat("0", 40) + "\n"
pr = scanRequest(signed(t, hash, "pull a b\n", "pub", "s"))
out, _ := f.uvAnswer(ctx, "r", pr, uvWho{user: "pub", from: pr.coveredFrom, nobodyReads: true}, 0)
if !strings.HasPrefix(string(out), "pragma uv-pull-only\n") {
t.Errorf("FAIL: a uv-hash before the login drew %q", out)
} else {
t.Log("ok: a uv-hash before the login card is answered as nobody")
}
out, _ = f.uvAnswer(ctx, "r", pr, uvWho{user: "pub", from: pr.coveredFrom, nobodyReads: false}, 0)
if len(out) != 0 {
t.Errorf("FAIL: nobody may not read, yet a uv-hash before the login drew %q", out)
} else {
t.Log("ok: where nobody may not read, a uv-hash before the login draws nothing")
}
}
// fossil reads a payload whenever SIZE is above 0 and no flag says
// otherwise, then refuses "-" as its hash.
func TestUVDeletionWithContentIsRefused(t *testing.T) {
f := New()
f.SetUnversioned(newFakeUV())
why := f.uvRefusal(context.Background(), "r", scanRequest([]byte("uvfile a.txt 9 - 5 0\nhello")), uvWho{user: "pub"})
if !strings.Contains(why, "HASH does not match CONTENT") {
t.Errorf("FAIL: refusal %q", why)
} else {
t.Log("ok: a deletion carrying content is refused")
}
}
// fossilReply is a child's compressed reply over body.
func fossilReply(body []byte) *http.Response {
return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"application/x-fossil"}},
Body: io.NopCloser(bytes.NewReader(encodeContainer(body)))}
}
// A write failing after one before it landed: the error card goes
// after the child's reply, which the client still gets, and the first
// write stays applied.
func TestUVWriteFailureIsAppended(t *testing.T) {
u := newFakeUV()
u.failOn = "b.txt"
f := New()
f.SetUnversioned(u)
body := "uvfile a.txt 9 " + uv.Hash([]byte("abc")) + " 3 0\nabc" +
"uvfile b.txt 9 " + uv.Hash([]byte("def")) + " 3 0\ndef" +
"uvfile c.txt 9 " + uv.Hash([]byte("ghi")) + " 3 0\nghi"
pr := scanRequest([]byte(body))
rec := httptest.NewRecorder()
if err := f.relayUV(context.Background(), rec, fossilReply([]byte("igot X\n")), "r", pr, uvWho{user: "pub"}); err != nil {
t.Fatal(err)
}
got := string(decodeContainer(rec.Body.Bytes()))
if !strings.HasPrefix(got, "igot X\nerror ") || !strings.Contains(got, "b.txt") {
t.Errorf("FAIL: reply = %q", got)
} else {
t.Log("ok: the child's reply arrives with the error after it")
}
if len(u.wrote) != 1 || u.wrote[0].Name != "a.txt" {
t.Errorf("FAIL: applied %+v", u.wrote)
} else {
t.Log("ok: the write before the failure stays, none after it is tried")
}
}
// A child reply too large to hold is streamed, and the cards still
// follow it, inside a container whose length counts them.
func TestUVLargeChildReplyIsStreamed(t *testing.T) {
u := newFakeUV()
f := New()
f.SetUnversioned(u)
child := bytes.Repeat([]byte("igot 0123456789abcdef0123456789abcdef01234567\n"), maxBufferedReply/46+100)
pr := scanRequest([]byte("pragma uv-hash " + strings.Repeat("0", 40) + "\nuvgimme hello.txt\n"))
rec := httptest.NewRecorder()
if err := f.relayUV(context.Background(), rec, fossilReply(child), "r", pr, uvWho{}); err != nil {
t.Fatal(err)
}
got, ok, _ := openContainer(rec.Body.Bytes(), maxDecoded)
tail := string(got[min(len(child), len(got)):])
hello := u.files[1]
if !ok || !bytes.Equal(got[:len(child)], child) || !strings.HasPrefix(tail, "pragma uv-pull-only\n") ||
!strings.HasSuffix(tail, "uvigot hello.txt 7 "+hello.Hash+" 6\n") {
t.Errorf("FAIL: ok=%v len=%d tail=%q", ok, len(got), tail)
} else {
t.Log("ok: a large reply streams through with the catalog after it, and the content deferred")
}
}
// A file over the bucket's current limit is answered with uvigot, as
// fossil answers a file it will not send this round.
func TestUVTooLargeIsDeferred(t *testing.T) {
u := newFakeUV()
u.tooLarge = map[string]bool{"hello.txt": true}
f := New()
f.SetUnversioned(u)
out, _ := f.uvAnswer(context.Background(), "r", scanRequest([]byte("uvgimme hello.txt\n")), uvWho{}, 0)
if want := "uvigot hello.txt 7 " + u.files[1].Hash + " 6\n"; string(out) != want {
t.Errorf("FAIL: %q", out)
} else {
t.Log("ok: a file over the limit is offered again, never sent wrong")
}
}
// A failure under the write lease is told apart from the lease itself
// refusing: only the second means this node is read-only.
func TestGuardFailureIsNotReadOnly(t *testing.T) {
if _, err := exec.LookPath("zit"); err != nil {
t.Skip("zit not on PATH")
}
origin := t.TempDir()
zrun(t, origin, "init")
code, err := ProjectCode(origin)
if err != nil {
t.Fatal(err)
}
f := New()
defer f.Close()
f.Register("alpha", origin)
f.SetUnversioned(newFakeUV())
mux := http.NewServeMux()
mux.Handle("POST /r/{repo}", f)
srv := httptest.NewServer(mux)
defer srv.Close()
round := func() string {
body := "push " + code + " " + code + "\npragma uv-hash " + strings.Repeat("0", 40) + "\n"
resp, err := http.Post(srv.URL+"/r/alpha", "application/x-fossil", bytes.NewReader(encodeContainer([]byte(body))))
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
raw, _ := io.ReadAll(resp.Body)
// One container and nothing after it: a second reply written
// after the first would trail it.
r := bytes.NewReader(raw[min(4, len(raw)):])
zr, err := zlib.NewReader(r)
if err != nil {
return "not a container: " + string(raw)
}
got, _ := io.ReadAll(zr)
if r.Len() != 0 {
return fmt.Sprintf("%s\n+%d trailing bytes: %q", got, r.Len(), decodeContainer(raw[len(raw)-r.Len():]))
}
return string(got)
}
f.SetGuard(func(_ string, proxy func() error) error {
proxy()
return errors.New("the relay failed after it answered")
})
if got := round(); strings.Contains(got, `read-only`) || !strings.Contains(got, "pragma uv-pull-only") {
t.Errorf("FAIL: a failure under the lease answered %q", got)
} else {
t.Log("ok: a failure under the lease keeps the reply it started and is not called read-only")
}
f.SetGuard(func(string, func() error) error { return errors.New("not the lease holder") })
if got := round(); !strings.Contains(got, `read-only`) {
t.Errorf("FAIL: a refused lease answered %q", got)
} else {
t.Log("ok: a refused lease is read-only")
}
}
// announced is a compressed container whose length says n whatever it
// holds.
func announced(n uint32, body []byte) []byte {
c := encodeContainer(body)
binary.BigEndian.PutUint32(c[:4], n)
return c
}
func childReply(raw []byte, ctype string) *http.Response {
return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {ctype}},
Body: io.NopCloser(bytes.NewReader(raw))}
}
// A large reply is checked to its end before any unversioned write is
// applied: zit appends some errors after output it has already served.
// A reply that does not hold what it announced, or announces more than
// the front adds cards to, goes to the client as the child sent it, and
// no write is applied either.
func TestLargeRepliesAreCheckedBeforeWriting(t *testing.T) {
big := bytes.Repeat([]byte("igot 0123456789abcdef0123456789abcdef01234567\n"), maxBufferedReply/46+100)
write := "uvfile a.txt 9 " + uv.Hash([]byte("abc")) + " 3 0\nabc"
pr := scanRequest([]byte(write))
refused := append(append([]byte{}, big...), "error peer\\sversion\\stoo\\sold\n"...)
for name, c := range map[string]struct {
reply []byte
ctype string
verbatim bool
}{
"an error at the end, compressed": {encodeContainer(refused), "application/x-fossil", true},
"an error at the end, uncompressed": {refused, "application/x-fossil-uncompressed", true},
"shorter than announced": {announced(uint32(len(big)+10), big), "application/x-fossil", true},
"past what the front adds cards to": {announced(maxStreamedReply+1, big), "application/x-fossil", true},
"a clean large reply": {encodeContainer(big), "application/x-fossil", false},
} {
u := newFakeUV()
f := New()
f.SetUnversioned(u)
rec := httptest.NewRecorder()
if err := f.relayUV(context.Background(), rec, childReply(c.reply, c.ctype), "r", pr, uvWho{user: "pub"}); err != nil {
t.Fatalf("%s: %v", name, err)
}
same := bytes.Equal(rec.Body.Bytes(), c.reply)
switch {
case c.verbatim && (len(u.wrote) != 0 || !same):
t.Errorf("FAIL: %s: applied %d writes, relayed as sent %v", name, len(u.wrote), same)
case !c.verbatim && len(u.wrote) != 1:
t.Errorf("FAIL: %s: applied %d writes, want 1", name, len(u.wrote))
default:
t.Logf("ok: %s: %d writes applied, relayed as sent %v", name, len(u.wrote), same)
}
}
}