// unversioned.go is RFC 0020 rung 3: the front answers fossil's
// unversioned cards itself, from the repository's bucket, and hands the
// zit child only the DAG part of a request.
//
// Everything here follows fossil's server, src/xfer.c at trunk:
// page_xfer's handling of `pragma uv-hash`, `uvgimme` and `uvfile`,
// send_unversioned_catalog, send_unversioned_file and
// xfer_accept_unversioned_file.
package xfer
import (
"bufio"
"bytes"
"compress/zlib"
"context"
"encoding/binary"
"errors"
"fmt"
"io"
"log"
"net/http"
"os"
"strconv"
"strings"
"craftdesign.group/quarry/internal/uv"
)
// Unversioned is a repository's unversioned files as the front needs
// them. The composition root supplies it from the catalog and the
// bucket; nil leaves the front as it was, where an unversioned card
// fails closed and the child refuses it.
type Unversioned interface {
// UVCatalog is the catalog, tombstones included. bound is false
// for a repository with no bucket of its own, which has no
// unversioned files.
UVCatalog(ctx context.Context, repo string) (files []uv.File, bound bool, err error)
// UVWritable is the binding's per-file limit, or why no unversioned
// write can land in repo at all (no bucket, a private repository
// whose bucket is public), in words for the refusal.
UVWritable(ctx context.Context, repo string) (limit int64, err error)
// UVContent is f's bytes from the bucket, read whole and bounded;
// a file over the binding's current limit is ErrUVTooLarge.
UVContent(ctx context.Context, repo string, f uv.File) ([]byte, error)
// UVWrite applies a version by fossil's rule, content or tombstone.
UVWrite(ctx context.Context, repo string, f uv.File, content []byte) (uv.File, bool, error)
// UVTouch moves a file's mtime when the hash is the one held: a
// uvfile that omits its content because only the mtime changed.
UVTouch(ctx context.Context, repo string, f uv.File) (uv.File, bool, error)
// MayPublish reports whether user holds y on repo.
MayPublish(repo, user string) bool
}
// SetUnversioned installs the unversioned-file answerer.
func (f *Fronter) SetUnversioned(u Unversioned) { f.uv = u }
// uvReplyBudget is fossil's max-download default (xfer.mxSend on the
// server): once the reply has reached it, send_unversioned_file sends
// a uvigot instead of the content, and the client asks again on its
// next round. The child's reply counts toward it, as it would in
// fossil's one reply.
const uvReplyBudget = 5000000
// maxBufferedReply is the most of the child's reply the front holds
// in memory to append to it: fossil's budget and a margin. A larger
// reply is streamed through with the cards added after it (relayUV), so
// a clone -u round carrying many artifacts still gets its catalog.
const maxBufferedReply = uvReplyBudget + 1<<20
// ErrUVTooLarge is a file whose bytes are over the binding's current
// limit, which happens when the limit is lowered after it was
// published. UVContent wraps it.
var ErrUVTooLarge = errors.New("larger than this bucket's limit")
// uvWho is who each unversioned card speaks for. fossil's page_xfer
// reads a request in order, so a card before the login card runs with
// nobody's permissions. quarry's equivalent is the login's signature:
// it covers only the bytes after the login line, or the whole body for
// the x-f-l-c cookie form (RFC ยง6.5), so a card before from could have
// been put there by anybody holding a copy of the signed request.
type uvWho struct {
user string // the verified login, "" for none
from int // the first byte the login's signature covers
// nobodyReads is whether the read gate admits nobody, which is
// what a card outside the signed region is answered as.
nobodyReads bool
}
// of is who card c speaks for and whether that party may read.
func (w uvWho) of(c uvCard) (user string, mayRead bool) {
if w.user != "" && c.start >= w.from {
return w.user, true
}
// With no login at all the request passed the read gate as nobody.
return "", w.user == "" || w.nobodyReads
}
// uvRefusal checks, before the child sees anything, every uvfile card a
// request carries, and answers fossil's error text for the first that
// cannot land. A refusal refuses the whole request, the DAG part with
// it: fossil's page_xfer stops at the first bad card and resets the
// reply, and quarry goes further and never applies the rest.
func (f *Fronter) uvRefusal(ctx context.Context, repo string, pr pushRequest, who uvWho) string {
if !pr.uvWrite {
return ""
}
if f.writeGate != nil {
if ok, upstream := f.writeGate(repo); !ok {
return "not accepted here: this store mirrors " + upstream + " and does not accept local writes, unversioned ones included"
}
}
// xfer_accept_unversioned_file's own words first, then what to do.
const missing = "Write permissions for unversioned files missing"
limit := int64(-1)
for _, c := range pr.uv {
if c.kind != "uvfile" {
continue
}
in, why := uvFileCard(c)
if why != "" {
return why
}
switch user, _ := who.of(c); {
case user == "" && who.user != "":
return missing + ": " + in.Name + " comes before the login card, which does not sign it"
case user == "":
return missing + ": publishing unversioned files needs a login holding y (Write-UV); sync with --user"
case !f.uv.MayPublish(repo, user):
return missing + ": " + user + " does not hold y (Write-UV) here, which a repository admin grants under Settings, Access; admin does not imply it"
}
if limit < 0 {
var err error
if limit, err = f.uv.UVWritable(ctx, repo); err != nil {
return err.Error()
}
}
if in.Deleted || c.omitted() {
continue
}
if int64(len(c.payload)) > limit {
return fmt.Sprintf("%s is %d bytes, over this bucket's limit of %d", in.Name, len(c.payload), limit)
}
if !uv.Verify(in.Hash, c.payload) {
return "in uvfile line, HASH does not match CONTENT: " + in.Name
}
}
return ""
}
// uvFileCard reads a framed uvfile card as the file it offers, or
// fossil's words for why it is malformed. The checks are
// xfer_accept_unversioned_file's, with quarry's own name rules.
func uvFileCard(c uvCard) (uv.File, string) {
name, mtimeTok, hash := c.f[1], c.f[2], c.f[3]
if err := uv.CheckName(name); err != nil {
return uv.File{}, "malformed uvfile line: " + err.Error()
}
mtime, err := strconv.ParseInt(mtimeTok, 10, 64)
if err != nil || mtime < 0 {
return uv.File{}, "malformed uvfile line: " + name + " has no mtime"
}
size, _ := strconv.ParseInt(c.f[4], 10, 64)
if hash == "-" {
// A deletion carries no content. fossil reads a payload
// whenever SIZE is above 0 and neither flag 0x0001 nor 0x0004
// is set, then finds "-" is not its hash.
if c.payload != nil {
return uv.File{}, "in uvfile line, HASH does not match CONTENT: " + name
}
return uv.File{Name: name, Mtime: mtime, Deleted: true}, ""
}
if !isHname(hash) {
return uv.File{}, "malformed uvfile line: " + name + " has no hash"
}
return uv.File{Name: name, Mtime: mtime, Hash: hash, Size: size}, ""
}
// isHname is fossil's hname test: a SHA1 or SHA3-256 hash in lower-case
// hex.
func isHname(s string) bool {
if len(s) != 40 && len(s) != 64 {
return false
}
for i := 0; i < len(s); i++ {
if (s[i] < '0' || s[i] > '9') && (s[i] < 'a' || s[i] > 'f') {
return false
}
}
return true
}
// uvAnswer applies a request's unversioned cards in order and returns
// the cards to append to the reply. It runs after the child answered
// the DAG part without refusing it, so a refused push leaves no
// unversioned write behind either. replySize is the child's reply,
// which counts toward fossil's reply budget.
//
// An error is a write that failed after the checks passed (the bucket
// or the catalog). The writes before the failing card are applied and
// stay applied; the failing one and every card after it are not, and
// out holds what was answered up to it. The check-ins the child took
// stay taken too: they are in the store before the bucket is asked,
// and nothing undoes them. The caller appends the error to the child's
// reply, so the client sees both, and its next sync -u offers the
// unversioned files again. Only refusals, which all come before the
// child, refuse the whole request.
func (f *Fronter) uvAnswer(ctx context.Context, repo string, pr pushRequest, who uvWho, replySize int) ([]byte, error) {
var out bytes.Buffer
var files map[string]uv.File
catalog := func() (map[string]uv.File, []uv.File, bool, error) {
list, bound, err := f.uv.UVCatalog(ctx, repo)
if err != nil || !bound {
return nil, nil, bound, err
}
m := make(map[string]uv.File, len(list))
for _, x := range list {
m[x.Name] = x
}
return m, list, true, nil
}
catalogSent := false
for _, c := range pr.uv {
user, mayRead := who.of(c)
switch c.kind {
case "uvfile":
in, _ := uvFileCard(c) // checked in uvRefusal
in.Who = user
var err error
switch {
case in.Deleted:
_, _, err = f.uv.UVWrite(ctx, repo, in, nil)
case c.omitted():
// Flag 0x0004: the content was left out. A stock client
// does that only for a file whose hash the server already
// holds with an older mtime (uv_tosend.mtimeOnly), and
// fossil's server then updates the mtime alone. A
// different hash without content would leave fossil a row
// with no bytes; with no bytes to put in the bucket,
// quarry applies nothing and the next uv-hash round
// offers the client the catalog again.
_, _, err = f.uv.UVTouch(ctx, repo, in)
default:
_, _, err = f.uv.UVWrite(ctx, repo, in, c.payload)
}
if err != nil {
return out.Bytes(), fmt.Errorf("unversioned file %s not stored, nor any after it: %w", in.Name, err)
}
files = nil // the catalog changed
case "uv-hash":
// page_xfer answers the first uv-hash only, only for a
// reader, and only when the hashes differ. A repository
// with no bucket answers nothing, which is what a fossil
// server says to a user who may not see the files.
if catalogSent || len(c.f) < 3 || !isHname(c.f[2]) {
continue
}
catalogSent = true
if !mayRead {
continue
}
m, list, bound, err := catalog()
if err != nil {
log.Printf("xfer: %s: unversioned catalog: %v", repo, err)
continue
}
if !bound || c.f[2] == uv.CatalogHash(list) {
continue
}
files = m
if f.uvPushOK(repo, user) {
out.WriteString("pragma uv-push-ok\n")
} else {
out.WriteString("pragma uv-pull-only\n")
}
// send_unversioned_catalog: a tombstone goes as hash "-"
// and size 0.
for _, x := range list {
hash, size := x.Hash, x.Size
if x.Deleted || hash == "" {
hash, size = "-", 0
}
fmt.Fprintf(&out, "uvigot %s %d %s %d\n", x.Name, x.Mtime, hash, size)
}
case "uvgimme":
if !mayRead {
continue
}
if files == nil {
m, _, bound, err := catalog()
if err != nil {
log.Printf("xfer: %s: unversioned catalog: %v", repo, err)
continue
}
if !bound {
continue
}
files = m
}
x, ok := files[c.f[1]]
if !ok {
continue // fossil's SELECT finds no row and sends nothing
}
f.sendUVFile(ctx, repo, &out, x, replySize)
}
}
return out.Bytes(), nil
}
// sendUVFile is send_unversioned_file on the server: a tombstone as
// "- 0 1"; once the reply has reached the budget, a uvigot so the
// client asks again next round; otherwise the content.
//
// A file that cannot be read is answered with that same uvigot. It is
// fossil's answer for a file too large to send now, and the client
// asks again on its next round, rather than a uvfile it would store
// wrongly or silence that leaves it waiting.
func (f *Fronter) sendUVFile(ctx context.Context, repo string, out *bytes.Buffer, x uv.File, replySize int) {
if x.Deleted || x.Hash == "" {
fmt.Fprintf(out, "uvfile %s %d - 0 1\n", x.Name, x.Mtime)
return
}
deferred := func() { fmt.Fprintf(out, "uvigot %s %d %s %d\n", x.Name, x.Mtime, x.Hash, x.Size) }
if replySize+out.Len() >= uvReplyBudget {
deferred()
return
}
content, err := f.uv.UVContent(ctx, repo, x)
switch {
case errors.Is(err, ErrUVTooLarge):
// Every round asks for it again; once a name is enough.
if _, seen := f.uvTooLarge.LoadOrStore(repo+"\x00"+x.Name, true); !seen {
log.Printf("xfer: %s: unversioned file %s is not served: %v", repo, x.Name, err)
}
deferred()
return
case err != nil:
log.Printf("xfer: %s: unversioned file %s: %v", repo, x.Name, err)
deferred()
return
case !uv.Verify(x.Hash, content):
log.Printf("xfer: %s: unversioned file %s: the bucket's bytes are not the catalog's hash; run uv.rebuild", repo, x.Name)
deferred()
return
}
fmt.Fprintf(out, "uvfile %s %d %s %d 0\n", x.Name, x.Mtime, x.Hash, len(content))
out.Write(content)
}
// uvPushOK is page_xfer's choice between uv-push-ok and uv-pull-only:
// a verified login holding y, on a store that takes local writes. A
// tracking mirror is pull-only for everyone.
func (f *Fronter) uvPushOK(repo, login string) bool {
if login == "" || !f.uv.MayPublish(repo, login) {
return false
}
if f.writeGate != nil {
if ok, _ := f.writeGate(repo); !ok {
return false
}
}
return true
}
// uvTail is what goes after the child's reply: the front's cards, and
// the error that stopped them, if one did.
func (f *Fronter) uvTail(ctx context.Context, repo string, pr pushRequest, who uvWho, replySize int) []byte {
tail, err := f.uvAnswer(ctx, repo, pr, who, replySize)
if err != nil {
log.Printf("xfer: %s: %v", repo, err)
tail = append(tail, errorCard(err.Error())...)
}
return tail
}
// maxStreamedReply is the largest child reply the front adds cards to.
// A larger one is relayed as the child sent it, without them, and the
// client's next sync -u asks again. It keeps a container's length, the
// child's plus the cards', far inside the four bytes that carry it.
const maxStreamedReply = maxRequest
// relayUV relays the child's reply with the front's unversioned cards
// appended, which fossil's message format allows: a reply is a
// sequence of cards, and a client reads uvigot and uvfile wherever they
// fall.
//
// Whatever its size, the reply is read to its end and checked for an
// error card before any unversioned write is applied. zit puts some
// errors at the END of a reply it has otherwise served (the peer-version
// gate and private cards, serverErrorReset's "append" sites in
// zit/src/sync.zig), so a large reply is no proof of success. The check
// is firstError's, line by line: a payload line that reads like an
// error card withholds the writes, which is the safe way to be wrong,
// and the client's next sync -u offers them again.
//
// A reply up to maxBufferedReply is held in memory. A larger compressed
// one is spooled to a temporary file while it is checked, because its
// length goes first and has to count the cards: they are worked out
// once the check has passed, and the child's body is recompressed from
// the spool with the cards after it. Past fossil's budget a uvgimme is
// answered with uvigot, so the cards stay small. A larger uncompressed
// one carries no length and is relayed as it arrives, checked on the
// way, with the cards written after it.
func (f *Fronter) relayUV(ctx context.Context, w http.ResponseWriter, resp *http.Response, repo string, pr pushRequest, who uvWho) error {
br := bufio.NewReaderSize(resp.Body, 64<<10)
if ct := resp.Header.Get("Content-Type"); ct != "" {
w.Header().Set("Content-Type", ct)
}
if resp.StatusCode != http.StatusOK {
w.WriteHeader(resp.StatusCode)
_, err := io.Copy(w, br)
return err
}
if n, ok := containerLength(br); ok && !strings.Contains(resp.Header.Get("Content-Type"), "uncompressed") && n > maxBufferedReply {
return f.relayUVStream(ctx, w, br, n, repo, pr, who)
}
reply, err := io.ReadAll(io.LimitReader(br, maxBufferedReply+1))
if err != nil {
return err
}
if len(reply) > maxBufferedReply {
w.WriteHeader(resp.StatusCode)
scan := &errorScan{}
out := io.MultiWriter(w, scan)
if _, err := out.Write(reply); err != nil {
return err
}
rest, err := io.Copy(out, br)
if err != nil {
return err
}
if scan.found {
log.Printf("xfer: %s: child refused the sync at the end of a large reply; unversioned cards withheld", repo)
return nil
}
f.landed(repo, pr, resp.StatusCode)
_, err = w.Write(f.uvTail(ctx, repo, pr, who, len(reply)+int(rest)))
return err
}
body, compressed, _ := openContainer(reply, maxDecoded)
if msg := firstError(body); msg != "" {
log.Printf("xfer: %s: child refused the sync: %s", repo, msg)
w.WriteHeader(resp.StatusCode)
_, err := w.Write(reply)
return err
}
f.landed(repo, pr, resp.StatusCode)
merged := reply
if tail := f.uvTail(ctx, repo, pr, who, len(body)); len(tail) > 0 {
merged = append(body, tail...)
if compressed {
merged = encodeContainer(merged)
}
}
w.WriteHeader(resp.StatusCode)
_, err = w.Write(merged)
return err
}
// relayUVStream is relayUV for a compressed reply announcing n bytes,
// too large to hold. The compressed bytes are spooled while their
// content is checked; only a reply that holds exactly n bytes and no
// error card gets its unversioned cards, and anything else goes to the
// client exactly as the child sent it, with no unversioned write
// applied.
func (f *Fronter) relayUVStream(ctx context.Context, w http.ResponseWriter, br *bufio.Reader, n int, repo string, pr pushRequest, who uvWho) error {
var hdr [4]byte
if _, err := io.ReadFull(br, hdr[:]); err != nil {
return err
}
verbatim := func(from io.Reader) error {
w.WriteHeader(http.StatusOK)
if _, err := w.Write(hdr[:]); err != nil {
return err
}
_, err := io.Copy(w, from)
return err
}
if n > maxStreamedReply {
log.Printf("xfer: %s: child reply of %d bytes is past %d; relayed without unversioned cards, none applied", repo, n, maxStreamedReply)
return verbatim(br)
}
spool, err := os.CreateTemp("", "quarry-xfer-reply-*")
if err != nil {
return err
}
defer os.Remove(spool.Name())
defer spool.Close()
scan := &errorScan{}
held, readErr := int64(0), error(nil)
zr, err := zlib.NewReader(io.TeeReader(br, spool))
if err == nil {
held, readErr = io.Copy(scan, io.LimitReader(zr, int64(n)+1))
zr.Close()
} else {
readErr = err
}
// Whatever follows the zlib stream belongs to the reply too.
if _, err := io.Copy(spool, br); err != nil {
return err
}
if _, err := spool.Seek(0, io.SeekStart); err != nil {
return err
}
switch {
case readErr != nil:
log.Printf("xfer: %s: child reply does not decompress (%v); relayed as sent, no unversioned write applied", repo, readErr)
return verbatim(spool)
case held != int64(n):
log.Printf("xfer: %s: child reply announced %d bytes and held %d; relayed as sent, no unversioned write applied", repo, n, held)
return verbatim(spool)
case scan.found:
log.Printf("xfer: %s: child refused the sync at the end of a large reply; unversioned cards withheld", repo)
return verbatim(spool)
}
f.landed(repo, pr, http.StatusOK)
tail := f.uvTail(ctx, repo, pr, who, n)
binary.BigEndian.PutUint32(hdr[:], uint32(n+len(tail)))
w.WriteHeader(http.StatusOK)
if _, err := w.Write(hdr[:]); err != nil {
return err
}
zr, err = zlib.NewReader(spool)
if err != nil {
return err
}
defer zr.Close()
zw := zlib.NewWriter(w)
if _, err := io.Copy(zw, io.LimitReader(zr, int64(n))); err != nil {
// The writes are applied and the reply is half sent; the
// client sees a broken reply and its next sync -u reconciles.
log.Printf("xfer: %s: relaying the child's reply after unversioned writes were applied: %v", repo, err)
return err
}
if _, err := zw.Write(tail); err != nil {
return err
}
return zw.Close()
}
// errorScan finds a line beginning "error ", firstError's test, in a
// stream, without holding more of it than one line's first six bytes.
type errorScan struct {
head []byte
mid bool // inside a line, past its first six bytes
found bool
}
func (e *errorScan) Write(p []byte) (int, error) {
for _, c := range p {
switch {
case c == '\n':
e.head, e.mid = e.head[:0], false
case e.mid:
default:
e.head = append(e.head, c)
if len(e.head) == 6 {
e.found = e.found || string(e.head) == "error "
e.mid = true
}
}
}
return len(p), nil
}
// containerLength peeks at a reply for a compressed container's
// announced length: four bytes of length and then a zlib header, which
// no card line begins with.
func containerLength(br *bufio.Reader) (int, bool) {
b, err := br.Peek(6)
if err != nil || b[4]&0x0f != 8 || (int(b[4])<<8|int(b[5]))%31 != 0 {
return 0, false
}
return int(binary.BigEndian.Uint32(b[:4])), true
}