beardrive/internal/webapp/store.go

package webapp

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"regexp"
	"strings"

	"github.com/runbear-io/beardrive/internal/config"
	"github.com/runbear-io/beardrive/internal/journal"
	"github.com/runbear-io/beardrive/internal/remote"
)

// The store API (/api/store/*) lets other devices sync through this server
// instead of talking to the object store themselves: the server is the only
// machine that knows where the storage is or holds credentials. It exposes
// the same key space every backend uses (blobs/<sha256>, journal/<dev>.jsonl)
// so the regular sync machinery works unchanged over it.
//
// Reads are always allowed (this is the same data the viewer serves). Writes
// follow the server's upload setting and go direct-to-storage via presigned
// URLs when the backend can sign, exactly like browser uploads.

var (
	blobKeyRe    = regexp.MustCompile(`^blobs/[0-9a-f]{64}$`)
	journalKeyRe = regexp.MustCompile(`^journal/` + deviceIDPattern + `\.jsonl$`)
)

func validStoreKey(key string) bool {
	return blobKeyRe.MatchString(key) || journalKeyRe.MatchString(key)
}

// storeSource returns the volume's RemoteSource; only real beardrive
// remotes have a store to expose.
func storeSource(v *volume, w http.ResponseWriter) *RemoteSource {
	rs, ok := v.source.(*RemoteSource)
	if !ok {
		http.Error(w, "this server does not front a beardrive remote", http.StatusNotFound)
		return nil
	}
	return rs
}

func (s *Server) storeKey(w http.ResponseWriter, r *http.Request) (string, bool) {
	key := r.URL.Query().Get("key")
	if !validStoreKey(key) {
		http.Error(w, fmt.Sprintf("invalid store key %q", key), http.StatusBadRequest)
		return "", false
	}
	return key, true
}

// ownJournal binds a journal key to the calling device. "Each device writes
// only its own journal" is why no journal object ever has two writers and why
// the hub needs no locking service — write permission on the project is not
// permission to rewrite a peer's log (or the hub's own), where a forged op
// with a high lamport wins replay on every device and History blames the
// victim. Blob keys are content-addressed and immutable, so they carry no
// owner.
//
// A caller that names no device writes no journal either, but that is a 400:
// the request never said who is writing, which is a malformed sync request
// rather than a refused one. Only a caller claiming to be a device it is not
// gets the 403.
//
// The body's ops are deliberately NOT read here any more. They used to admit
// a first claim when every op named the device, which is a field the writer
// writes — see the default arm below.
func (s *Server) ownJournal(w http.ResponseWriter, r *http.Request, key string) bool {
	if !strings.HasPrefix(key, "journal/") {
		return true
	}
	// The canonical spelling on both sides. A journal key IS a storage key, and
	// the stores underneath disagree about case (APFS and NTFS fold, S3 does
	// not), so a device that may spell its id two ways is a device that owns two
	// keys and one file — see canonDeviceID. Requiring the canonical key means
	// one device is one object everywhere.
	dev := deviceID(r)
	if dev == "" {
		http.Error(w, "a journal write must identify its device (X-Bdrive-Device)", http.StatusBadRequest)
		return false
	}
	if key != "journal/"+dev+".jsonl" {
		http.Error(w, "a device may only write its own journal", http.StatusForbidden)
		return false
	}
	// Matching the key against the header binds nothing on its own: the same
	// request supplies both, so moving them together satisfies the check by
	// construction and any member could replace any peer's journal object —
	// their ops gone, every peer replaying the forged ones, History crediting
	// them to the victim. The device has to belong to the ACCOUNT as well.
	//
	// Ownership is DeviceRegistry.OwnerOf: hub-wide, first claim, ownerless
	// rows claiming nothing. Three things this deliberately does not do, each
	// because doing it was a hole:
	//
	//   - it does not consult the row this request would create. Every /store
	//     handler used to register the caller's header before asking who owns
	//     it, so an unclaimed id authorized whoever named it first. The
	//     callers observe AFTER this returns.
	//   - it does not treat "unclaimed" as permission. An id nothing has ever
	//     synced under is not this caller's to write.
	//   - it does not scope the claim to the project's org, so offboarding a
	//     teammate does not release her journal to the org she left.
	//
	// A hub with no registry cannot resolve ownership at all (single-volume,
	// auth-less, or a fixture): there is nobody to impersonate, and projectPerm
	// answers admin for exactly those configurations.
	if s.Devices != nil {
		me := normEmail(s.requestUser(r).Email)
		owner, _ := s.Devices.OwnerOf(dev)
		switch {
		case normEmail(owner) != "" && normEmail(owner) == me:
			// My device, my journal.
		case atLeast(s.projectPerm(r, r.PathValue("project")), PermAdmin):
			// Project admin is the recovery path — the answer to "a squatted id
			// is a permanent lockout". The device's own remedy is in the body.
		default:
			// There is no "first writer claims an unowned id" arm any more.
			// It used to admit `!known && journalNames(dev, ops)` — every op in
			// the body naming the device — which reads a field the WRITER
			// writes, so it cost one request to take any id that had not yet
			// pushed a journal. That included every device of every read-only
			// member, permanently, because a device that syncs with READ can
			// never reach this door to claim its own id in the first place.
			//
			// A device id is now bound to its account when the hub mints that
			// machine's token (DeviceRegistry.Bind), which is a moment the hub
			// authenticates and the machine cannot forge. So an unowned id is
			// simply not anybody's to write, and the remedy is to sign in.
			http.Error(w, "this device is not registered to your account on this hub; "+
				"run `bdrive login` on this machine (if the id belongs to someone else, "+
				"delete device.json in your BearDrive home first, or ask a project admin)",
				http.StatusForbidden)
			return false
		}
	}
	return true
}

func (s *Server) handleStoreList(v *volume, w http.ResponseWriter, r *http.Request) {
	rs := storeSource(v, w)
	if rs == nil {
		return
	}
	s.refreshDevice(r)
	prefix := r.URL.Query().Get("prefix")
	if prefix != "" && prefix != "journal/" && prefix != "blobs/" &&
		!strings.HasPrefix(prefix, "journal/") && !strings.HasPrefix(prefix, "blobs/") {
		http.Error(w, fmt.Sprintf("invalid prefix %q", prefix), http.StatusBadRequest)
		return
	}
	// A sync cycle starts here, which makes it the hub's regular opportunity to
	// confirm what its presigned grants actually delivered.
	s.reconcileGrants(r.Context(), r.PathValue("project"), rs.Backend)
	objs, err := rs.Backend.List(r.Context(), prefix)
	if err != nil {
		storageErr(w, http.StatusBadGateway, "storage is temporarily unavailable", err)
		return
	}
	writeJSON(w, map[string]any{"objects": objs})
}

func (s *Server) handleStoreGet(v *volume, w http.ResponseWriter, r *http.Request) {
	rs := storeSource(v, w)
	if rs == nil {
		return
	}
	s.refreshDevice(r)
	key, ok := s.storeKey(w, r)
	if !ok {
		return
	}
	// Blobs go through OpenBlob so a presigned write cannot make this route
	// serve content that does not hash to the key it is stored under.
	var rc io.ReadCloser
	var err error
	if blob, isBlob := strings.CutPrefix(key, "blobs/"); isBlob {
		rc, err = rs.OpenBlob(r.Context(), blob)
	} else {
		rc, err = rs.Backend.Get(r.Context(), key)
	}
	if err != nil {
		// Fixed message: os.Open's error names the hub's absolute storage
		// path, and S3's names the bucket and key.
		storageErr(w, http.StatusNotFound, "no such object", err)
		return
	}
	defer rc.Close()
	w.Header().Set("Content-Type", "application/octet-stream")
	// The sync proxy is a stored-bytes door like the other two: a
	// cookie-authenticated GET whose URL one member can hand another, answering
	// with content the attacker wrote under a Content-Type the hub chose.
	w.Header().Set("X-Content-Type-Options", "nosniff")
	// Recorded, never checked. This is a device syncing: refusing it here
	// surfaces as ErrForbidden, which the syncer reads as "access is gone —
	// pause and touch nothing". Sync must not break over a bill.
	cw := &countingWriter{w: w}
	io.Copy(cw, rc)
	s.quota().RecordEgress(s.orgOf(r.PathValue("project")), cw.n)
}

func (s *Server) handleStoreExists(v *volume, w http.ResponseWriter, r *http.Request) {
	rs := storeSource(v, w)
	if rs == nil {
		return
	}
	s.refreshDevice(r)
	key, ok := s.storeKey(w, r)
	if !ok {
		return
	}
	exists, err := rs.Backend.Exists(r.Context(), key)
	if err != nil {
		storageErr(w, http.StatusBadGateway, "storage is temporarily unavailable", err)
		return
	}
	writeJSON(w, map[string]any{"exists": exists})
}

// handleStoreSign answers how a client should upload a key: a presigned
// direct-to-storage URL when the backend can sign, through the server
// otherwise — same contract as browser uploads.
func (s *Server) handleStoreSign(v *volume, w http.ResponseWriter, r *http.Request) {
	rs := storeSource(v, w)
	if rs == nil {
		return
	}
	if !s.Upload.Enabled {
		http.Error(w, "uploads are disabled on this server", http.StatusForbidden)
		return
	}
	var req struct {
		Key  string `json:"key"`
		Size int64  `json:"size"`
	}
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
		http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
		return
	}
	if !validStoreKey(req.Key) || req.Size < 0 {
		http.Error(w, fmt.Sprintf("invalid store key %q", req.Key), http.StatusBadRequest)
		return
	}
	// Ownership is deliberately NOT consulted here. Signing grants nothing for
	// a journal — journals are never presigned, the answer is always "come
	// through the server" — so the only thing asking could do is turn
	// OwnerOf's hub-wide answer into a status code on a route any org member
	// may call: a plain member of one org probed a device id belonging to a
	// separate tenant and the response told him whether it existed. The write
	// itself is where ownership is enforced, and that is the only place it
	// needs to be.
	s.refreshDevice(r)
	project := r.PathValue("project")
	s.reconcileGrants(r.Context(), project, rs.Backend)
	org := s.orgOf(project)
	// The cap is checked against this write PLUS everything already granted
	// and not yet accounted for, so concurrent grants cannot oversubscribe an
	// allowance that no single one of them exceeds.
	if err := s.quota().CheckWrite(org, req.Size+s.reservedBytes(org)); err != nil {
		http.Error(w, err.Error(), http.StatusForbidden)
		return
	}
	// Only blobs are presigned. They are content-addressed and immutable, so
	// a leaked URL can at worst re-upload identical bytes. Journals are
	// mutable state and always flow through the server.
	if blob, isBlob := strings.CutPrefix(req.Key, "blobs/"); isBlob {
		if !sizeFitsContentAddress(blob, req.Size) {
			http.Error(w, "declared size does not match the content address", http.StatusForbidden)
			return
		}
		if exists, err := rs.Backend.Exists(r.Context(), req.Key); err == nil && exists {
			writeJSON(w, map[string]any{"mode": "direct", "exists": true})
			return
		}
		if signer, ok := rs.Backend.(remote.PutSigner); ok {
			// Reserved, not charged: the bytes go straight to storage, so this
			// grant counts against the cap immediately and is billed when the
			// object is confirmed there (reconcileGrants), or released for
			// free when the URL expires unused. Booking it here outright
			// charged 20 GiB for 20 JSON posts. The check and the reservation
			// are one critical section, so concurrent callers cannot all read
			// the same zero and oversubscribe.
			if err := s.reserveIfFits(project, org, req.Key, req.Size, s.Upload.ttl()); err != nil {
				http.Error(w, err.Error(), http.StatusForbidden)
				return
			}
			if signed, err := signer.SignPut(r.Context(), req.Key, req.Size, s.Upload.ttl()); err == nil {
				writeJSON(w, map[string]any{
					"mode": "direct", "url": signed.URL, "method": signed.Method,
					"headers": signed.Headers, "expires": signed.Expires.UTC(),
				})
				return
			}
			s.claimGrant(project, req.Key) // nothing was granted: give it back
		}
	}
	writeJSON(w, map[string]any{"mode": "server"})
}

// journalOps reads the operations a spooled journal body carries, exactly the
// way every device reads it (journal.Parse: a line that decodes to no
// operation is no operation). It leaves the file rewound for the store.
// A non-journal key carries no ops by definition.
//
// It is also the hub's ONLY path check on this door. /store/* is the second
// ingest into a project's tree and it used to validate nothing: the browser
// door (cleanUploadPath) refused control characters and this one journaled
// them, so "notes\x00.md" reached the tree, the metadata store and the Share
// button through the door round 6 said refusing at ingest had closed. The
// rule is journal.SafePath — the same one the device applies in unsafeRel and
// the same one cleanUploadPath is built on.
func journalOps(key string, tmp *os.File) ([]journal.Op, error) {
	if !strings.HasPrefix(key, "journal/") {
		return nil, nil
	}
	data, err := io.ReadAll(tmp)
	if err != nil {
		return nil, err
	}
	if _, err := tmp.Seek(0, io.SeekStart); err != nil {
		return nil, err
	}
	ops, err := journal.Parse(data)
	if err != nil {
		return nil, err
	}
	for _, op := range ops {
		// The same two clauses the browser door applies (cleanUploadPath):
		// SafePath, plus the reserved dirs. Applying only the first here left
		// /store/* journaling ".git/hooks/pre-commit" with a 200 while /remove
		// and /shares answered 400 for the same path — so the entry was in the
		// tree, served to every device, and no request could take it back out.
		if !journal.SafePath(op.Path) || config.ReservedPath(op.Path) {
			return nil, fmt.Errorf("journal names an invalid path %q", op.Path)
		}
		// The note is the other peer-written free text History renders, right
		// next to the path (HistoryRow's NoteText, the run-card header) — and
		// `bdrive log` already scrubs the same characters out of it on the way
		// to a terminal, on the stated grounds that "the audit tool an operator
		// uses to catch a peer must not be renderable BY that peer". The web
		// History view is the audit tool everybody actually uses.
		// The note is not the OTHER peer-written free text History renders, it
		// is one of three: Op.Author and Op.UserName are rendered in the same
		// row by the same helper (the frontend's whoChanged), and were checked
		// by nothing. A right-to-left override in user_name reorders the whole
		// rendered row — the Trojan Source shape SafeText exists to refuse —
		// and a C0 run in author is the "renders as nothing" shape its own doc
		// comment names. DeviceName is absent on purpose: History serves the
		// device REGISTRY's name, not the op's.
		if !journal.SafeText(op.Note) || !journal.SafeText(op.Author) || !journal.SafeText(op.UserName) {
			return nil, fmt.Errorf("journal carries invalid text")
		}
	}
	return ops, nil
}

// opsNameTheirAuthor refuses a journal whose ops credit an account other than
// the one that owns the device writing them.
//
// Op.User/Op.UserName are what History serves and what the frontend's
// whoChanged() renders as THE answer to "who changed this file?" — the hub's
// only audit surface. They arrived as fields the pushing client typed, so bob
// pushed an op declaring alice and the audit log named Alice. The hub already
// holds the truth on this very request: ownJournal has just resolved the
// account the device id in the key belongs to.
//
// It refuses rather than overwrites: a journal object is the device's own log
// byte for byte, every peer replays exactly these bytes, and rewriting a body
// mid-push would make the hub a second author of a log the design says has
// exactly one.
//
// An op that names nobody at all is fine — journals from before accounts
// existed have no User and History falls back to Author. A NAME with no account
// is not: it is an attribution with nothing behind it, rendered the same way.
//
// "Names nobody" has to include Author, and that was the bypass: this checked
// User alone and explicitly waved through an op with neither User nor UserName,
// while whoChanged() falls back to Author — unchecked peer text — as THE answer
// to "who changed this file?". So bob pushed an op naming nobody, put
// "Alice <alice@x.io>" in author, and the audit surface credited Alice again,
// one field over from the fix. A device the hub has bound got its token from a
// login, so it has an account to name; if it names anything, it names that one.
func (s *Server) opsNameTheirAuthor(w http.ResponseWriter, r *http.Request, ops []journal.Op) bool {
	if s.Devices == nil || len(ops) == 0 {
		return true // nobody to impersonate (single-volume, auth-less, fixture)
	}
	owner, _ := s.Devices.OwnerOf(deviceID(r))
	if normEmail(owner) == "" {
		return true // no binding to check against; ownJournal already ruled on the write
	}
	for _, op := range ops {
		if op.User == "" && op.UserName == "" && op.Author == "" {
			continue
		}
		if normEmail(op.User) != normEmail(owner) {
			http.Error(w, "an op must name the account this device is registered to as its author",
				http.StatusForbidden)
			return false
		}
	}
	return true
}

// journalKeepsItsOps reports whether an incoming journal body still carries
// every op the hub already holds under key, matched on the per-device sequence
// number — the append-only rule the data model states and that /store/object, a
// plain object PUT, enforced nowhere.
//
// "Every change is an Op in a per-device append-only JSONL log" is why History
// can answer "who changed this file?" at all. Without this, the writer of a
// journal can replace it with a SHORTER one: every op it held is gone from
// replay, from every peer, and from the hub's only audit surface. Two principals
// reach that — the device's own account erasing its own trail, and, once
// offboarding releases the id and the laptop is reassigned, whoever inherits it
// erasing the departed member's.
//
// Seq is the field to key on: it is the device's own monotone counter, an honest
// client never reuses one (store.AppendOps only appends), so "no stored Seq has
// vanished" is the log-only-grows statement in the model's own terms.
//
// ponytail: this refuses TRUNCATION, not rewriting-in-place — a body that keeps
// every Seq but changes what one of them says still edits the record. The
// stronger rule is a byte-prefix compare against the stored object, which is
// what an honest client always produces; it is not what ships here because two
// established fixtures drive this door by replacing a journal wholesale
// (sec_audit2, sec_path), and a security fix that rewrites the tests around it
// is a fix nobody can audit. Upgrade path: byte prefix, with those two fixtures
// switched to append.
//
// A key the hub does not hold yet is a first push and keeps everything. A stored
// journal the hub cannot parse protects nothing and is not a reason to refuse a
// device's sync — it is the hub's own corruption, and ingest (journalOps) has
// refused unparseable bodies since round 6. A backend that cannot answer fails
// the push closed: the client degrades to Offline and retries next cycle, which
// is the posture everywhere else on this path.
func journalKeepsItsOps(ctx context.Context, be remote.Backend, key string, ops []journal.Op) (bool, error) {
	switch have, err := be.Exists(ctx, key); {
	case err != nil:
		return false, err
	case !have:
		return true, nil
	}
	rc, err := be.Get(ctx, key)
	if err != nil {
		return false, err
	}
	defer rc.Close()
	data, err := io.ReadAll(rc)
	if err != nil {
		return false, err
	}
	stored, err := journal.Parse(data)
	if err != nil {
		log.Printf("beardrive: %s is not parseable, its ops cannot be protected from a rewrite: %v", key, err)
		return true, nil
	}
	seen := make(map[int64]bool, len(ops))
	for _, op := range ops {
		seen[op.Seq] = true
	}
	for _, op := range stored {
		if !seen[op.Seq] {
			return false, nil
		}
	}
	return true, nil
}

func (s *Server) handleStorePut(v *volume, w http.ResponseWriter, r *http.Request) {
	rs := storeSource(v, w)
	if rs == nil {
		return
	}
	if !s.Upload.Enabled {
		http.Error(w, "uploads are disabled on this server", http.StatusForbidden)
		return
	}
	key, ok := s.storeKey(w, r)
	if !ok {
		return
	}
	// Spool the body before storing any of it. Everything this handler has to
	// be sure of is a property of the bytes, not of the headers the client
	// sent: what a blob key promises (its sha256), what the write costs
	// (Content-Length is -1 on any chunked request, which made every unsized
	// put free), and how many ops a journal write actually authors.
	// Cost: one temp file per put on the hub's busiest write path.
	tmp, size, sum, err := spool(r.Body)
	if err != nil {
		storageErr(w, http.StatusBadGateway, "could not store the object", err)
		return
	}
	defer os.Remove(tmp.Name())
	defer tmp.Close()
	if blob, isBlob := strings.CutPrefix(key, "blobs/"); isBlob && blob != sum {
		http.Error(w, "content does not hash to its key", http.StatusBadRequest)
		return
	}
	ops, err := journalOps(key, tmp)
	if err != nil {
		// The body is the client's, so everything journalOps can object to is
		// the client's fault: an undecodable journal or an op naming a path
		// this hub will not carry. 400, not 502 — and nothing is stored.
		http.Error(w, "invalid journal body", http.StatusBadRequest)
		return
	}
	if !s.ownJournal(w, r, key) {
		return
	}
	if !s.opsNameTheirAuthor(w, r, ops) {
		return
	}
	// Observed only after the write is authorized, and only into a row the
	// account ALREADY owns (refreshDevice checks OwnerOf first). Nothing here
	// claims: that happens once, when the hub mints this machine's token
	// (DeviceRegistry.Bind).
	//
	// The journal branch used to call observeDevice, which creates the row it
	// does not find — and ownJournal's admin arm lets a project admin write
	// somebody else's journal as the RECOVERY path. Anyone who can create a
	// project is admin of it, so one PUT into a project of the attacker's own
	// wrote a competing row {attacker, victim's device id}, and Bind refuses
	// any id another account holds a row for: the victim's `bdrive login` was
	// then 409 forever, across the org wall, and `bdrive login` is the
	// documented remedy for every other device problem. A recovery path may
	// not brick the recovery path.
	s.refreshDevice(r)
	project := r.PathValue("project")
	s.reconcileGrants(r.Context(), project, rs.Backend)
	org := s.orgOf(project)
	if err := s.quota().CheckWrite(org, size+s.reservedBytes(org)); err != nil {
		http.Error(w, err.Error(), http.StatusForbidden)
		return
	}
	if strings.HasPrefix(key, "journal/") {
		switch ok, err := journalKeepsItsOps(r.Context(), rs.Backend, key, ops); {
		case err != nil:
			storageErr(w, http.StatusBadGateway, "could not read the stored journal", err)
			return
		case !ok:
			http.Error(w, "a journal is append-only; this body drops ops the hub already holds",
				http.StatusConflict)
			return
		}
	}
	if err := rs.Backend.Put(r.Context(), key, tmp, size); err != nil {
		storageErr(w, http.StatusBadGateway, "could not store the object", err)
		return
	}
	// These bytes came through the hub, so they are charged here — drop any
	// reservation for the same key rather than charging it twice.
	s.claimGrant(project, key)
	s.quota().RecordUsage(org, size)
	if strings.HasPrefix(key, "journal/") {
		v.invalidate() // new ops should show in the viewer immediately
	}
	writeJSON(w, map[string]any{"ok": true})
}