// Package xfer fronts the zit sync protocol (0001 §7): quarry
// terminates HTTP for POST /r/{repo} — the URL zit clients use verbatim
// — and proxies each request to a per-repo `zit serve` child. TLS,
// keep-alive, and N-repos-on-one-port live here; zit's deliberately
// minimal one-shot server stays untouched. Children are started lazily
// and restarted if they die; a proxied push lands artifacts in the
// store, where the ingest watcher picks them up on its next poll.
package xfer
import (
"bytes"
"context"
"fmt"
"io"
"log"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/children"
)
// maxRequest caps a buffered sync request. Clients batch rounds well
// below this; the cap only bounds the gate's memory.
const maxRequest = 256 << 20
// Auth is the push-gating credential source (0001 §7: auth lives at
// quarry's edge). Gated reports whether a repo requires authenticated
// pushes; Secrets returns every shared secret (RFC §6.3) a login card
// for user may be signed with, in the order they are tried.
type Auth interface {
Gated(repo string) bool
Secrets(repo, user string) []Credential
// MayWrite reports the CHECK-IN LETTER alone, which Secret cannot
// say because it answers false for a person who has the letter and
// no credential AND for a person who has a credential and not the
// letter. Two different problems with two different remedies, and
// the refusal used to give one remedy for both — telling somebody
// who already holds a credential to mint another (jva's G-128).
MayWrite(repo, user string) bool
}
// Credential is one secret a login card may verify against, and what a
// request signed with it may write. An account holds several when it
// holds several sync tokens (RFC 0021), one per session, and a push
// signed with any of them lands.
type Credential struct {
Secret string
// Token is the id of the token the secret is derived from, "" for
// a hand-issued sync_user row.
Token string
// CheckIn is the i letter: false means this credential signs
// unversioned pushes only, and a request carrying a check-in card
// is refused whole.
CheckIn bool
// Publish is the y letter as this credential resolves it, which a
// token's ceiling can take away from an account that holds it.
Publish bool
}
// signedBy is the first credential login verifies against.
func signedBy(login pushRequest, creds []Credential) (Credential, bool) {
for _, c := range creds {
if verifyLogin(login, c.Secret) {
return c, true
}
}
return Credential{}, false
}
// ReadGate answers whether this sync request may read the repository at
// all. nil leaves every repository readable, which is v1's behaviour.
//
// It exists because the web gate is decoration without it: a private
// repository whose pages refuse anonymous readers, while `zit clone`
// against the same name hands over every artifact, is not private.
//
// login is the account named by a VERIFIED sync login card, or "" when
// the request carried none. It is the protocol's own credential, and
// accepting it here is what makes a private repository clonable by the
// people it admits: HTTP Basic reaches this gate, but a sync client
// does not send Basic — fossil and zit authenticate with the login
// card. Without it "private" meant "nobody may clone this, including
// its owner" (b112b205).
type ReadGate func(req *http.Request, repo, login string) bool
// WriteGuard runs a store-mutating proxy under the store's single-writer
// lease (0001 §3); it returns an error when this node may not write.
type WriteGuard func(repo string, proxy func() error) error
// Protect answers, for one authenticated pusher, which protected-branch
// globs they do NOT own (RFC 0007 §Enforcement). The gate turns an
// identity into a rule; the child turns the rule into a decision,
// because only the store can say which branch a pushed check-in
// resolves to. nil: nothing is protected.
type Protect func(repo, user string) ([]string, error)
// ProtectHeader carries the per-request rule to the child. Honoured by
// zit serve only when started with --fronted, which every child here
// is; the value is a comma-separated glob list, the shape zit's
// ignore-glob lists already use. Never on the fossil wire: loopback,
// between quarry and its own child.
const ProtectHeader = "X-Zit-Protect"
// Fronter proxies sync requests to per-repo zit serve children.
type Fronter struct {
mu sync.Mutex
repos map[string]*backend
budget *decodeBudget // decoded request bytes in flight
auth Auth // nil: every push is anonymous
limit LoginLimit // nil: login card misses are not counted
guard WriteGuard // nil: pushes are not lease-gated (single node)
readGate ReadGate // nil: every repository is readable
writeGate WriteGate // nil: every repository accepts pushes
pushed Pushed // nil: nobody is told a push landed
counted Counted // nil: nothing is tallied
identify Identify // nil: readers are never named
protect Protect // nil: no branch is protected
known Known // nil: a missing backend is always "no such repository"
uv Unversioned // nil: an unversioned card fails closed, as before RFC 0020
// uvTooLarge remembers the files already logged as over their
// bucket's limit, so a client asking every round is logged once.
uvTooLarge sync.Map
}
// Reader is what a served read said about who was asking. Every field
// is best-effort and most of them are empty most of the time: a public
// repository is mostly cloned by strangers, and that is not a defect in
// the capture.
//
// The three identity states are kept APART on purpose and must never be
// merged by a caller. Account is proven — a credential this forge
// checked. Claimed is a name the client asserted and nothing verified;
// nothing signs a pull, so a claim is free to forge. Neither set means
// nobody said who they were. Folding a claim in with an account would
// make a forged name indistinguishable from a real one at exactly the
// moment somebody reads the total.
type Reader struct {
// Clone distinguishes a new clone from a later fetch round.
Clone bool
// Client is the software, from the User-Agent: "zit", "fossil", or
// "other". Self-reported, so it answers "what is still on the old
// path" and never "who is this".
Client string
// Version is the client-version pragma's sequence. Also
// self-reported, and empty when the client sent no pragma.
Version string
// Account is a VERIFIED identity: a Basic credential this forge
// authenticated, or a login card whose signature checked out
// against the shared secret. Empty when nobody proved anything.
Account string
// Claimed is a login-card name whose signature did NOT check out,
// or could not be checked. Never treat it as an identity.
Claimed string
}
// Anonymous reports a read that named nobody at all.
func (r Reader) Anonymous() bool { return r.Account == "" && r.Claimed == "" }
// Counted is told that a repository was cloned or fetched, and what the
// request said about the reader. Stated in primitives rather than the
// events package's types for the same reason every other seam here is:
// the fronter knows what the protocol said, and the composition root
// owns what that is worth recording.
//
// It is called only for requests that were actually SERVED — after the
// visibility gate, after auth, and after the backend answered. A refused
// pull is not a fetch, and a repository that answered 404 to a stranger
// must not tell that stranger's arrival to a counter either.
type Counted func(repo string, r Reader)
// SetCounted installs the clone/fetch tally.
func (f *Fronter) SetCounted(c Counted) { f.counted = c }
// Pushed is told each time a round carrying a push was admitted by
// this forge AND answered by the store without a refusal — the one
// moment quarry knows a tracking store took writes its upstream never
// saw (5d40f151). The pushed check-ins are not named: the round is a
// container the child ingests, and what it contained is the fold
// loop's to read afterwards.
type Pushed func(repo string)
// SetPushed installs the push listener.
func (f *Fronter) SetPushed(p Pushed) { f.pushed = p }
// Known reports whether the forge serves a repository by this name at
// all, independently of whether its backend has been registered yet.
//
// The two answers differ for tens of seconds after a restart: stores
// are registered one at a time in name order, and a large forge takes a
// while to walk the list. A sync arriving in that window found no
// backend and was told the repository does not exist -- so a clone or
// push during a deploy reported a missing repository rather than a
// forge that was not ready yet, and the client had no reason to retry.
type Known func(name string) bool
// SetKnown installs the registry lookup that tells "not yet" from "no".
func (f *Fronter) SetKnown(k Known) { f.known = k }
// Identify resolves the VERIFIED account behind a request's Basic
// credential, or "" for none. It is the same answer the visibility gate
// already computes for the same request — deliberately, so that "who
// may read this" and "who is recorded as having read it" can never
// drift apart into two different notions of identity.
//
// It must never challenge and never refuse: it observes, it does not
// gate. A wrong password here costs a name off a statistic, never a
// clone.
type Identify func(req *http.Request) string
// LoginLimit is the forge's sign-in limiter, seen from the sync lane: a
// login card that verifies against none of the account's credentials
// is a miss against that account, and an account the limiter holds
// proves nothing with a card until the hold lapses (RFC 0021, Rate
// limits). The request is passed so the limiter can tell addresses
// apart by its own rule.
type LoginLimit interface {
Allowed(req *http.Request, user string) bool
Failed(req *http.Request, user string)
Succeeded(req *http.Request, user string)
}
// SetLoginLimit installs the limiter login card misses feed.
func (f *Fronter) SetLoginLimit(l LoginLimit) { f.limit = l }
// SetIdentify installs the reader-identity resolver for tallies.
func (f *Fronter) SetIdentify(i Identify) { f.identify = i }
// WriteGate answers whether the repository accepts pushes at all: a
// mirror that tracks an upstream normally does not, whoever is asking.
// The string names the upstream for the refusal. nil: every store is
// writable.
type WriteGate func(repo string) (ok bool, upstream string)
// SetWriteGate installs the posture gate for pushes.
func (f *Fronter) SetWriteGate(g WriteGate) { f.writeGate = g }
// SetAuth installs the push gate.
func (f *Fronter) SetAuth(a Auth) { f.auth = a }
// SetProtect installs the branch-protection rule source.
func (f *Fronter) SetProtect(p Protect) { f.protect = p }
// SetReadGate installs the visibility gate for pulls.
func (f *Fronter) SetReadGate(g ReadGate) { f.readGate = g }
// SetGuard installs the single-writer lease gate for pushes.
func (f *Fronter) SetGuard(g WriteGuard) { f.guard = g }
type backend struct {
// name is the repository as the forge calls it, carried so the
// child's log lines can say which store they came from.
name string
dir string
port int
cmd *exec.Cmd
// errR is our end of the child's stderr pipe, owned here rather
// than by os/exec. See startChild for why that distinction is the
// difference between a forge that stops and one that wedges.
errR *os.File
// exited closes once the reaper has waited for cmd. The reaper owns
// cmd's Wait; os/exec does not support a second, concurrent Wait on
// the same command, so this is how stop waits for the exit.
exited chan struct{}
}
// stop kills the child and releases its stderr reader.
//
// The order matters. Wait first: with an *os.File on Stderr it reaps
// the process and returns, because os/exec has no copier to await.
// Then close our read end, which unblocks the logging goroutine even
// if some grandchild still holds the write end open forever.
func (b *backend) stop() {
if b.cmd != nil && b.cmd.Process != nil {
b.cmd.Process.Kill()
<-b.exited
}
if b.errR != nil {
b.errR.Close()
b.errR = nil
}
b.cmd = nil
}
func New() *Fronter {
return &Fronter{repos: map[string]*backend{}, budget: requestBudget}
}
// decodes is the budget this front's decoded requests draw on: the
// process's own unless a test gave the front one of its own.
func (f *Fronter) decodes() *decodeBudget {
if f.budget == nil {
return requestBudget
}
return f.budget
}
// Register makes one repository frontable. dir is the repository
// directory (the one holding .zit), same as the ingest side uses.
func (f *Fronter) Register(name, dir string) {
f.mu.Lock()
defer f.mu.Unlock()
f.repos[name] = &backend{name: name, dir: dir}
}
// Unregister removes a repository and stops its child. Removing the
// entry without killing the child would leave a `zit serve` process
// holding a store nothing routes to any more — invisible, and still
// holding file handles.
func (f *Fronter) Unregister(name string) {
f.mu.Lock()
defer f.mu.Unlock()
b, ok := f.repos[name]
if !ok {
return
}
b.stop()
delete(f.repos, name)
}
// Close stops every child.
func (f *Fronter) Close() {
f.mu.Lock()
defer f.mu.Unlock()
for _, b := range f.repos {
b.stop()
}
}
// ServeHTTP handles POST /r/{repo} (and its /xfer alias): ensure the
// repo's child is alive, then relay the request body and response
// verbatim — the payload is the sync protocol's business, not ours.
func (f *Fronter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
name := req.PathValue("repo")
// The body is read BEFORE the visibility gate, because the sync
// protocol carries its credential inside it: the login card is how
// fossil and zit say who they are, and a gate that cannot see it
// can only ever admit anonymous readers. It is read once here and
// handed on untouched.
//
// The cost is buffering a request for a repository we may refuse.
// That is the same cost any public repository already pays, and it
// is bounded by maxRequest either way.
raw, err := io.ReadAll(io.LimitReader(req.Body, maxRequest+1))
if err != nil || len(raw) > maxRequest {
http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge)
return
}
// The client says how large its container inflates, and zlib reaches
// about a thousand to one, so the claim is checked before a byte is
// inflated. A claim past maxDecoded is refused rather than relayed:
// the child would inflate it just the same.
claim, _ := containerClaim(raw)
if int64(claim) > maxDecoded {
http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge)
return
}
// When a gate needs the cards, the body is inflated once, here, and
// every reader below shares the result. The claim is reserved against
// the process's decode budget first; once inflated, only the bytes
// actually decoded stay reserved, and those only until the relay to
// the child starts, so a slow reader of the reply holds nothing.
//
// The budget counts DECODED bytes. The raw body, up to maxRequest per
// request, is buffered outside it, as it always was: reserving it
// would mean holding the budget for the length of a slow upload.
body, compressed := raw, false
var pr pushRequest
budget := f.decodes()
var held int64
defer func() { budget.give(held) }()
if f.auth != nil || f.guard != nil || f.writeGate != nil || f.counted != nil || f.readGate != nil || f.pushed != nil || f.uv != nil {
if !budget.take(int64(claim)) {
log.Printf("xfer: %s: decode budget exhausted, %d bytes refused", name, claim)
w.Header().Set("Retry-After", "5")
http.Error(w, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable)
return
}
held = int64(claim)
body, compressed, err = openContainer(raw, maxDecoded)
if err != nil {
http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge)
return
}
decoded := int64(0)
if compressed {
decoded = int64(len(body))
}
budget.give(held - decoded)
held = decoded
// Inspect cards once: the read gate wants the login, auth
// verifies it on pushes, the lease routes pushes to the
// write-holder.
pr = scanRequest(body)
}
// Visibility first, and answering NOT FOUND rather than forbidden:
// a sync client told 403 has learned the repository exists, which is
// the one thing a private repository must not tell it.
//
// But a caller who offered NOTHING is challenged instead, because a
// bare 404 is a dead end for every real client. fossil's first
// clone request carries no credential at all: it sends Basic only
// after a 401, and a login card only on later rounds of a sync
// already under way. Answering 404 meant it was never challenged,
// never authenticated, and never reached round two — so a private
// repository was unclonable by the people it admits, however
// correctly the gate would have judged them (b112b205).
//
// The challenge cannot leak existence, because an UNKNOWN
// repository answers a credential-less caller exactly the same way
// — see the ensure branch below. Private and absent are one answer
// until somebody proves who they are, which is the property 404 was
// chosen for in the first place.
st := f.verifiedLogin(name, req, body, pr)
login := st.user
if f.readGate != nil && !f.readGate(req, name, login) {
f.refuseRead(w, req, st)
return
}
b, err := f.ensure(name)
if err != nil {
status := http.StatusBadGateway
// Logged AFTER the status is decided, so the line says what the
// client was actually told. It used to read "unknown repository"
// even while answering "starting", which is the misleading
// signal this whole change exists to remove -- and it cost real
// time reading it during a deploy.
reason := err.Error()
if err == errUnknownRepo {
status = http.StatusNotFound
// An unknown repository owes a credential-less caller the
// same answer a private one gives, or the difference
// between 401 and 404 becomes an existence oracle.
//
// Only where a read gate exists, though. A forge with no
// visibility policy has no private repositories to hide, so
// a name it does not serve is simply not found — and
// challenging for it would ask a bare server's clients to
// log in to something that never refuses anybody.
if f.readGate != nil && (f.known == nil || !f.known(name)) {
if f.challenges(req, st.user) {
f.refuseRead(w, req, st)
return
}
}
// A repository the forge serves but has not registered yet
// is starting, not missing. Say so with a retry hint: a
// clone or push arriving during a deploy should wait rather
// than conclude the repository is gone.
//
// This cannot leak the existence of a private repository:
// the visibility gate above already answered 404 for anyone
// who may not see it, so only a reader who is allowed to
// know the name reaches here at all.
if f.known != nil && f.known(name) {
w.Header().Set("Retry-After", "5")
status = http.StatusServiceUnavailable
reason = "not registered yet; the forge is still starting"
}
}
log.Printf("xfer: %s: %s (%d)", name, reason, status)
http.Error(w, http.StatusText(status), status)
return
}
// Posture before credentials: a mirror refuses the push whoever
// signed it, and the message says where the push belongs.
if pr.hasPush && f.writeGate != nil {
if ok, upstream := f.writeGate(name); !ok {
log.Printf("xfer: %s: push refused, store mirrors %s", name, upstream)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("not accepted here: this store mirrors " + upstream + " and does not accept local writes"))
return
}
}
// pusher is the AUTHENTICATED identity — the login card verified
// against the shared secret — and never the U card, which is a
// string any clone can write. Empty means anonymous, and anonymous
// owns nothing.
pusher := ""
if pr.hasPush && f.auth != nil && f.auth.Gated(name) {
// Body login when present; else the x-f-l-c cookie form
// (RFC §6.5), whose nonce covers the whole body.
login := pr
if login.user == "" {
login, _ = cookieLogin(req, body)
} else if pr.firstWrite >= 0 && pr.firstWrite < pr.coveredFrom {
// A body login signs only the bytes after its line, and a
// write card in front of it is nobody's: fossil reads the
// cards in order and runs those before the login as nobody.
// Credited to the login, it would let anyone holding one
// signed request (a pull will do) put artifacts of their
// own in front of it and replay it under that name. The
// cookie form signs the whole body and never gets here.
log.Printf("xfer: %s: push rejected (user %q): a write card comes before the login card", name, login.user)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("not authorized to write: a push or file card comes before the login card, which does not sign it"))
return
}
// The limiter was consulted, and a miss counted, by
// verifiedLogin above: the same card, so it is not counted twice.
if st.held {
log.Printf("xfer: %s: push rejected (user %q): held by the sign-in limiter", name, login.user)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("not authorized to write as " + login.user +
": too many failed sign-ins for this account; wait and try again"))
return
}
creds := f.auth.Secrets(name, login.user)
known := len(creds) > 0
cred, verified := signedBy(login, creds)
if !verified {
log.Printf("xfer: %s: push rejected (user %q)", name, login.user)
w.Header().Set("Content-Type", "application/x-fossil")
// Name what is actually missing. "not authorized to write"
// sent two people straight to the capability letters, which
// are only half of the answer: a push needs the letters AND
// a credential, and the credential is the half with no page
// pointing at it. A refusal that does not say which half is
// wrong costs an afternoon.
w.Write(errorReply(refusalFor(login.user,
f.auth.MayWrite(name, login.user), known)))
return
}
// A credential holding y without i publishes unversioned files
// and nothing else: the request is refused whole the moment it
// carries a card the DAG would take.
if !cred.CheckIn && pr.dagWrite {
log.Printf("xfer: %s: push rejected (user %q): an unversioned-only credential carried a check-in card", name, login.user)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("not authorized to write as " + login.user +
": the credential presented publishes unversioned files only, and this push carries check-in content"))
return
}
pusher = login.user
}
// The per-request protection rule, computed for EVERY request and
// sent even when empty. zit#151 treats an absent header as no rule
// — right for a bare server, but behind a front it would let a
// forgotten code path silently disable protection. So "empty
// because computed" is the only shape that exists here; "absent
// because forgotten" cannot happen, and the harness proves it.
//
// A rule source that cannot answer fails the push closed: the one
// thing worse than refusing an owner is admitting a stranger
// because a database was busy.
rule := ""
if f.protect != nil {
refused, err := f.protect(name, pusher)
if err != nil {
if pr.hasPush {
log.Printf("xfer: %s: push refused, protection unavailable: %v", name, err)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("branch protection is unavailable; push refused"))
return
}
} else {
rule = strings.Join(refused, ",")
}
}
// RFC 0020: the unversioned cards are the front's to answer, from
// the catalog and the bucket, and the child is sent the request
// without them, compressed again if it came compressed. Every
// uvfile is checked here, before the child sees anything, so a
// refusal refuses the whole request; the writes themselves wait
// until the child has taken the DAG part.
//
// Cutting cards out of a signed request is only possible because
// the child holds no credential: zit serve's authFault accepts any
// login when it has none (zit/src/sync.zig, authFault, "const cred =
// credential orelse return null"), and runServe configures none.
// A child that verified the login itself would see a body whose
// nonce no longer matches and refuse every round that carried an
// unversioned card. The login is verified here, against the bytes
// the client signed, before anything is cut.
send := raw
uvOn := f.uv != nil && (len(pr.uv) > 0 || pr.uvBad != "")
var who uvWho
if uvOn {
who = uvWho{user: st.user, withheld: st.withheld, from: st.from, nobodyReads: f.readGate == nil || f.readGate(req, name, "")}
why := pr.uvBad
if why == "" {
why = f.uvRefusal(req.Context(), name, pr, who)
}
if why != "" {
log.Printf("xfer: %s: unversioned sync refused (%s): %s", name, st.why(), why)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply(why))
return
}
send = stripUV(body, pr.uv)
if compressed {
send = encodeContainer(send)
}
}
// The relay lasts as long as the client takes to read the reply, so
// the decoded body must be unreachable before it starts: a parsed
// login keeps only a digest of what it signs, the tally's login is
// taken now, and the unversioned payloads the writes still need after
// the relay are copied out, staying reserved, so that no slice keeps
// the whole body alive. A plaintext body is raw itself, which lives
// until the request ends whatever is done here.
var tallied pushRequest
if f.counted != nil {
tallied = requestLogin(req, pr, body)
}
if compressed {
var kept int64
for i, c := range pr.uv {
if c.payload != nil {
pr.uv[i].payload = bytes.Clone(c.payload)
kept += int64(len(c.payload))
}
}
budget.give(held - kept)
held = kept
}
body = nil
// rw says whether the relay has started a reply, so a failure after
// that point is not answered a second time.
rw := &startedWriter{ResponseWriter: w}
proxy := func() error {
target := "http://127.0.0.1:" + strconv.Itoa(b.port) + "/xfer"
prox, err := http.NewRequestWithContext(req.Context(), "POST", target, bytes.NewReader(send))
if err != nil {
return err
}
// A fresh request, not a copy: the client's headers never reach
// the child. That is what makes the protect header the gate's
// alone — a forwarded client header with that name would be a
// bypass on loopback behind --fronted.
prox.Header.Set("Content-Type", req.Header.Get("Content-Type"))
prox.Header.Set(ProtectHeader, rule)
resp, err := backendClient.Do(prox)
if err != nil {
// The child may have died mid-flight; drop it so the next
// request restarts it.
f.drop(name)
return err
}
defer resp.Body.Close()
if uvOn {
return f.relayUV(req.Context(), rw, resp, name, pr, who)
}
if ct := resp.Header.Get("Content-Type"); ct != "" {
rw.Header().Set("Content-Type", ct)
}
rw.WriteHeader(resp.StatusCode)
// A REFUSED SYNC is the one reply worth reading on the way
// past, because the refusal exists nowhere else.
//
// Giving the child a voice (childlog.go) assumed it would say
// why on stderr. It does not: `zit serve` reports a rejected
// card ON THE WIRE, as an error card, and says nothing on
// stderr at all. So a push that failed a dozen times left the
// forge log holding one line — the child's startup banner —
// and the operator still could not answer "why". That is the
// same gap the child's voice was meant to close, one layer out.
//
// Bounded by SIZE, not by whether the request parsed as a push.
// pr is only filled in when one of the gates is configured
// (see above), so keying this on pr.hasPush would have made
// the forge log depend on which unrelated planes happen to be
// wired — silent exactly where a bare relay needs it most. A
// served clone is megabytes and is nobody's error message, so
// anything past the cap streams through untouched and the cost
// on the path that matters stays zero.
head, err := io.ReadAll(io.LimitReader(resp.Body, maxRefusalPeek+1))
if err != nil {
return err
}
if _, werr := rw.Write(head); werr != nil {
return werr
}
if len(head) <= maxRefusalPeek {
// The whole peek is inflated, not a prefix of it, because the
// child puts some errors at the end of a reply (see relayUV);
// a peek inflating past the most the front holds of a reply
// is history, not a refusal.
peek, _, _ := openContainer(head, maxBufferedReply)
if msg := firstError(peek); msg != "" {
log.Printf("xfer: %s: child refused the sync: %s", name, msg)
return nil
}
f.landed(name, pr, resp.StatusCode)
return nil
}
f.landed(name, pr, resp.StatusCode)
_, err = io.Copy(rw, resp.Body)
return err
}
// A push mutates the store, so it routes through the write lease; a
// node that does not hold it answers with a read-only error card
// rather than writing behind the holder's back. An unversioned write
// changes the bucket and the catalog, and takes the same lease.
//
// The guard's error is either the lease's refusal, before proxy ran,
// or proxy's own failure under the lease; only the first means this
// node is read-only.
if (pr.hasPush || uvOn && pr.uvWrite) && f.guard != nil {
ran := false
if err := f.guard(name, func() error { ran = true; return proxy() }); err != nil {
if ran {
f.proxyFailed(rw, name, err)
return
}
log.Printf("xfer: %s: push not served here: %v", name, err)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("store is read-only on this node"))
}
return
}
if err := proxy(); err != nil {
f.proxyFailed(rw, name, err)
return
}
f.tally(name, pr, tallied, req)
}
// proxyFailed reports a relay that failed after it was admitted: a 502
// when no reply has started, and only the log line once one has, since
// the client is already reading it.
func (f *Fronter) proxyFailed(rw *startedWriter, name string, err error) {
log.Printf("xfer: %s: proxy: %v", name, err)
if !rw.started {
http.Error(rw, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
}
}
// startedWriter records whether a reply has begun.
type startedWriter struct {
http.ResponseWriter
started bool
}
func (s *startedWriter) WriteHeader(code int) {
s.started = true
s.ResponseWriter.WriteHeader(code)
}
func (s *startedWriter) Write(p []byte) (int, error) {
s.started = true
return s.ResponseWriter.Write(p)
}
// maxRefusalPeek bounds the reply the relay reads before passing it
// on. A refusal is one error card; anything at this size is history
// being served, which is streamed through untouched.
const maxRefusalPeek = 64 << 10
// refusalFor says which half of a push's authorization is missing, in
// the one sentence a fossil error card gets.
//
// A push needs TWO things: the check-in letter on this repository, and
// a push credential. The refusal used to name both and say one was
// missing, then prescribe minting a credential — so a person who held
// a credential and lacked the letter was sent to mint a second one,
// which changes nothing. jva measured all four cells and found that
// cell wrong (G-128): a wrong remedy is worse than a missing one,
// because it is followable.
//
// The place named for granting the letter was first written as "its
// people page", from reading the code rather than using the page. That
// page lists CONTRIBUTORS — names found on artifacts, not accounts — and
// has no role control; roles are granted under Settings, Access. jva
// checked it from a repository admin's seat (G-134). A remedy that names
// the wrong place is the same defect as a remedy that names the wrong
// cure, one step later.
//
// mayWrite is the letter; hasCredential is whether one exists at all.
// The third state is the one dmestas hit by accident: both present, and
// the card signed with the wrong secret — an account password where the
// sync token belongs. Nothing is missing there, so "missing" was the
// wrong word as well as the wrong cure.
func refusalFor(user string, mayWrite, hasCredential bool) string {
switch {
case user == "":
return "anonymous push is not accepted here: push with --user and a push credential"
case !mayWrite:
return "not authorized to write as " + user +
": this repository has not granted " + user + " the check-in letter," +
" which a repository admin grants under Settings, Access." +
" A push credential alone is not enough"
case !hasCredential:
return "not authorized to write as " + user +
": " + user + " may check in here but holds no push credential" +
" (mint one with POST /api/v1/me/sync-token)"
}
return "not authorized to write as " + user +
": the credential presented did not verify." +
" A push is signed with the sync token from POST /api/v1/me/sync-token," +
" not with an account password"
}
// loginState is what a request offered and what became of it.
//
// The bool alone was not enough to debug with: a refused read logged
// nothing, so "no card was sent", "a card was sent and did not verify"
// and "a verified account may not read this" were one silent 404 from
// outside. Three different bugs, one symptom, and four attempts at it
// from a client settled none of them.
type loginState struct {
user string // the verified account, "" when there is none
offered bool // a login card was present in the request
noSecret bool // no shared secret exists for that account here
basic bool // HTTP Basic credentials were present
// from is the first byte of the body the verified login's signature
// covers: after the login line, or 0 for the cookie form.
from int
// withheld is a verified credential that does not reach y.
withheld bool
// held is an account the sign-in limiter is holding: its card was
// not checked.
held bool
}
// why describes the credential situation in one short phrase, for a log
// line a person reads at three in the morning.
func (st loginState) why() string {
switch {
case st.user != "":
return "login card verified as " + st.user
case st.held:
return "login card offered while the account is held for too many failed sign-ins"
case st.offered && st.noSecret:
return "login card offered but this account has no push credential here"
case st.offered:
return "login card offered and did not verify"
case st.basic:
return "HTTP Basic offered, no login card"
}
return "no credential offered"
}
// verifiedLogin reports what this request PROVED it is, through the
// sync protocol's own credential.
//
// It never refuses anything: a bad card is the same as no card here,
// and the caller decides what that means. The push path verifies
// separately and REFUSES, because presenting a credential that does
// not check out is a different thing from presenting none.
func (f *Fronter) verifiedLogin(repo string, req *http.Request, body []byte, pr pushRequest) loginState {
_, _, hasBasic := req.BasicAuth()
st := loginState{basic: hasBasic}
if f.auth == nil {
return st
}
login := requestLogin(req, pr, body)
if login.user == "" {
return st
}
st.offered = true
creds := f.auth.Secrets(repo, login.user)
if len(creds) == 0 {
st.noSecret = true
return st
}
if f.limit != nil && !f.limit.Allowed(req, login.user) {
st.held = true
return st
}
cred, ok := signedBy(login, creds)
if !ok {
if f.limit != nil {
f.limit.Failed(req, login.user)
}
return st
}
if f.limit != nil {
f.limit.Succeeded(req, login.user)
}
st.user, st.from, st.withheld = login.user, login.coveredFrom, !cred.Publish
return st
}
// challenges reports whether this request offered no credential at all,
// and so should be invited to present one rather than told nothing is
// here.
func (f *Fronter) challenges(req *http.Request, login string) bool {
if login != "" {
return false
}
_, _, hasBasic := req.BasicAuth()
return !hasBasic
}
// refuseRead answers a read this forge will not serve.
//
// A caller who offered nothing is challenged, so a client that
// authenticates on demand — which is every stock sync client — gets the
// chance to. A caller who offered something and still may not read is
// told 404: they have had their turn, and repeating the challenge would
// loop. Both answers are identical for a repository that does not
// exist, so neither tells anybody whether it does.
func (f *Fronter) refuseRead(w http.ResponseWriter, req *http.Request, st loginState) {
repo := req.PathValue("repo")
if f.challenges(req, st.user) {
log.Printf("xfer: %s: read challenged (401): %s", repo, st.why())
w.Header().Set("WWW-Authenticate", `Basic realm="quarry"`)
http.Error(w, http.StatusText(http.StatusUnauthorized), http.StatusUnauthorized)
return
}
// The one a person actually needs: the caller presented SOMETHING
// and is being told nothing is here. Without this line the three
// reasons that lead here are indistinguishable from outside, which
// is exactly how a private repository stayed unclonable while every
// measurement said 404.
log.Printf("xfer: %s: read refused (404): %s", repo, st.why())
http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
}
// tally records a served read. Pushes are not counted here: they are
// already the timeline's business, where a clone leaves no other trace
// at all — which is the whole reason to count these two.
// landed tells the push listener a round with a push got through: the
// forge admitted it and the store answered it without an error card.
func (f *Fronter) landed(name string, pr pushRequest, status int) {
if f.pushed == nil || !pr.hasPush || status != http.StatusOK {
return
}
f.pushed(name)
}
// requestLogin is the request's login card: the one in its body, or
// else the cookie form (RFC §6.5), whose nonce covers the whole body.
func requestLogin(req *http.Request, pr pushRequest, body []byte) pushRequest {
if pr.user != "" {
return pr
}
login, _ := cookieLogin(req, body)
return login
}
func (f *Fronter) tally(name string, pr, login pushRequest, req *http.Request) {
switch {
case f.counted == nil:
case pr.newClone:
f.counted(name, f.reader(name, pr, login, req, true))
case pr.hasPull:
f.counted(name, f.reader(name, pr, login, req, false))
}
}
// reader assembles what this request said about who was asking.
//
// Nothing here can refuse anything: it runs AFTER the response has been
// written. That ordering is the safety property — a bug in identity
// resolution cannot turn into a refused clone, because by the time this
// runs the clone has already been served.
func (f *Fronter) reader(name string, pr, login pushRequest, req *http.Request, clone bool) Reader {
r := Reader{Clone: clone, Client: clientKind(req.UserAgent()), Version: pr.version}
// Basic first: it is the form zit uses, it is checked against the
// account database, and it arrives on the clone round itself.
if f.identify != nil {
r.Account = f.identify(req)
}
if r.Account != "" {
return r
}
// Else a login card, which is fossil's form. It arrives on later
// rounds rather than on the clone, so this mostly names fetches.
if login.user == "" {
return r
}
// The signature check is the whole difference between an account
// and a claim. It is the same verification pushes already run; the
// only thing new is running it on a request shape that skipped it.
if f.auth != nil {
if _, ok := signedBy(login, f.auth.Secrets(name, login.user)); ok {
r.Account = login.user
return r
}
}
r.Claimed = login.user
return r
}
// clientKind names the software from its User-Agent. zit presents a
// fossil-shaped client-version pragma for compatibility, so the pragma
// card cannot tell the two apart and the User-Agent is the only signal
// that does.
//
// Self-reported, and trivially spoofable. It is a population statistic,
// never evidence.
func clientKind(ua string) string {
switch {
case strings.HasPrefix(ua, "zig/"), strings.Contains(ua, "zit/"):
return "zit"
case strings.HasPrefix(ua, "Fossil/"), strings.HasPrefix(ua, "fossil/"):
return "fossil"
default:
return "other"
}
}
// backendClient has no overall timeout: sync rounds on big stores are
// legitimately long. Cancellation rides the inbound request context.
var backendClient = &http.Client{}
var errUnknownRepo = fmt.Errorf("unknown repository")
func (f *Fronter) drop(name string) {
f.mu.Lock()
defer f.mu.Unlock()
if b, ok := f.repos[name]; ok && b.cmd != nil {
if b.cmd.Process != nil {
b.cmd.Process.Kill()
<-b.exited
}
b.cmd = nil
}
}
// ensure returns a live backend for name, starting its child if needed.
func (f *Fronter) ensure(name string) (*backend, error) {
f.mu.Lock()
defer f.mu.Unlock()
b, ok := f.repos[name]
if !ok {
return nil, errUnknownRepo
}
if b.cmd != nil {
// A quick dial beats trusting the reaper alone: it may not have
// run yet for a just-dead child, and a hung child that no longer
// accepts deserves a restart too. exited is read rather than
// ProcessState, which the reaper writes with no lock of ours.
alive := true
select {
case <-b.exited:
alive = false
default:
}
if alive {
if c, err := net.DialTimeout("tcp4", "127.0.0.1:"+strconv.Itoa(b.port), 300*time.Millisecond); err == nil {
c.Close()
return b, nil
}
b.cmd.Process.Kill()
}
b.stop()
}
port, err := freePort()
if err != nil {
return nil, err
}
// --fronted: this child's only client is the gate on loopback, and
// it may honour the gate's per-request protect header. A zit serve
// reachable directly is never started this way.
//
// Built by the process's children, so a stopping forge waits for
// this child and logs its end rather than exiting over it.
cmd, untrack, err := children.Default.Command(context.Background(), b.name, "serve",
"zit", "serve", "--port", strconv.Itoa(port), "--fronted")
if err != nil {
return nil, fmt.Errorf("start zit serve: %w", err)
}
cmd.Dir = b.dir
// The child's diagnostics reach the forge log (childlog.go) — but
// through a pipe THIS code owns, never by handing os/exec a plain
// io.Writer.
//
// os/exec connects an *os.File to the child directly and copies
// anything else on a goroutine that Wait() blocks on until every
// write end of the pipe is closed. Killing `zit serve` does not
// close a write end some grandchild inherited, so Wait() never
// returned: Close() and Unregister() hold f.mu while they wait, and
// ensure() and ServeHTTP take that same mutex, so one removed
// repository wedged sync for every repository on the forge. It hung
// two CI runs at the ten-minute timeout before it was believed.
//
// With an *os.File there is no copier to await, the reading is ours
// to end, and stop() ends it by closing errR.
pr, pw, err := os.Pipe()
if err != nil {
untrack()
return nil, fmt.Errorf("stderr pipe for zit serve: %w", err)
}
cmd.Stderr = pw
if err := cmd.Start(); err != nil {
untrack()
pr.Close()
pw.Close()
return nil, fmt.Errorf("start zit serve: %w", err)
}
// The parent's write end goes now: only the child (and whatever it
// spawns) should hold one, or the reader would never see EOF even
// after everybody exits.
pw.Close()
b.errR = pr
go func(log io.Writer, r *os.File) {
io.Copy(log, r)
r.Close()
}(newChildLog(b.name), pr)
// Reap in the background so ProcessState is set when it exits.
exited := make(chan struct{})
go func() {
cmd.Wait()
untrack()
close(exited)
}()
if err := waitReady(port, 5*time.Second); err != nil {
cmd.Process.Kill()
<-exited
return nil, err
}
b.cmd, b.port, b.exited = cmd, port, exited
return b, nil
}
// ProjectCode reads a store's identity through `zit stat -R` (zit#85).
// It works on a BARE store — no checkout, no server, no port — which is
// what lets quarry serve a repository that is nothing but a .zit file.
//
// This replaced a clone handshake against a transient `zit serve` child:
// find a free port, spawn, poll for readiness, POST a container, scrape
// the push card. That cost a subprocess and a listening socket per
// repository at startup, and because serve is checkout-only it was the
// single thing keeping quarry from bare-store serving.
func ProjectCode(dir string) (string, error) {
// Registration runs after the listener binds, so a stop can land
// while this is running; built by the process's children so the
// stop waits for it. A refusal fails the command before it runs.
cmd, untrack, _ := children.Default.Command(context.Background(), filepath.Base(dir), "stat",
"zit", "stat", "-R", dir)
defer untrack()
var stderr bytes.Buffer
cmd.Stderr = &stderr
out, err := cmd.Output()
if err != nil {
return "", fmt.Errorf("zit stat -R %s: %w: %s", dir, err, strings.TrimSpace(stderr.String()))
}
for _, line := range strings.Split(string(out), "\n") {
if rest, ok := strings.CutPrefix(line, "project-code "); ok {
if code := strings.TrimSpace(rest); code != "" {
return code, nil
}
}
}
return "", ErrNoProjectCode
}
// freePort asks the kernel for an unused port. The listen-then-close
// window is racy in principle; waitReady failing surfaces a collision
// and the next request retries with a fresh port.
func freePort() (int, error) {
l, err := net.Listen("tcp4", "127.0.0.1:0")
if err != nil {
return 0, err
}
port := l.Addr().(*net.TCPAddr).Port
l.Close()
return port, nil
}
func waitReady(port int, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
c, err := net.DialTimeout("tcp4", "127.0.0.1:"+strconv.Itoa(port), 200*time.Millisecond)
if err == nil {
c.Close()
return nil
}
time.Sleep(50 * time.Millisecond)
}
return fmt.Errorf("zit serve on :%d not ready within %s", port, timeout)
}
// ReapOrphans kills stray `zit serve --fronted` children left by a
// previous quarry.
//
// The fronter starts one per repository on demand. On a graceful stop
// serve's teardown stops every one: the registrar's CloseAll calls
// Unregister per repository, and Close follows. That teardown does not
// run when the process dies instead of stopping (the stop watchdog, a
// second signal, a crash, a SIGKILL), and before quarry handled SIGTERM
// it never ran at all: one orphan per synced repository per restart,
// reparented to init and running forever. Twenty deploys in a day left
// twenty-four of them, the oldest eight hours old, each holding a port
// and its own memory. This reap at startup is the fallback for those
// cases.
//
// A freshly started quarry has no backends by definition, so any child
// carrying --fronted is a leftover. The flag is the marker: it is set
// only here, for children whose sole client is the loopback gate.
func ReapOrphans() int {
out, err := exec.Command("pgrep", "-f", "zit serve --port .* --fronted").Output()
if err != nil {
return 0 // no pgrep, or nothing matched: nothing to do either way
}
killed := 0
for _, line := range strings.Fields(string(out)) {
pid, perr := strconv.Atoi(line)
if perr != nil || pid <= 1 {
continue
}
p, ferr := os.FindProcess(pid)
if ferr != nil {
continue
}
if p.Kill() == nil {
killed++
}
}
return killed
}