// bucket.go is RFC 0016 Phase 2: the bucket a repository's uploads go
// to, bound by a steward and proved before it is kept.
//
// The forge never holds the bytes, so what it holds instead has to be
// right: a binding is written only after a presigned PUT and DELETE
// succeeded against it, and a steward who mistypes a key sees the
// refusal on the form rather than a broken upload a week later. The
// secret goes in and never comes back — no Out here has a field that
// could carry it, the same construction mirror.chatbridge uses.
package api
import (
"context"
"errors"
"fmt"
"log"
"net"
"net/http"
"net/url"
"slices"
"strings"
"time"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/s3"
)
// BucketStore is the binding table. Stated in the OLTP package's own
// row type because the row carries the unsealed secret and only one
// seam should ever see it: this one. nil means this forge binds no
// buckets, which is a forge without an OLTP.
type BucketStore interface {
BindBucket(repo string, b oltp.BucketBinding) error
Bucket(repo string) (oltp.BucketBinding, bool, error)
UnbindBucket(repo string) error
SetBucketProbe(repo string, at time.Time, probeErr string, corsOK bool) error
}
// probeTimeout bounds one probe. A bucket that takes longer than this
// to answer two tiny requests is not one uploads will work against.
const probeTimeout = 20 * time.Second
// BucketInfo is a binding as every caller sees it: no secret, by
// construction. Inherited names the parent whose bucket a companion
// store uses (RFC 0015); a companion cannot bind its own.
type BucketInfo struct {
Repo string `json:"repo"`
Bound bool `json:"bound"`
Endpoint string `json:"endpoint,omitempty"`
Bucket string `json:"bucket,omitempty"`
Region string `json:"region,omitempty"`
AccessKey string `json:"accessKey,omitempty"`
// PublicBase is the domain downloads are linked from when the
// bucket is public; empty means every download is presigned.
PublicBase string `json:"publicBase,omitempty"`
// Limit is the per-file cap in bytes for uploads through quarry.
Limit int64 `json:"limit,omitempty"`
// ProbedAt is the last probe's stamp and ProbeErr what it said
// when it failed; a stamp with no error is a binding that works.
ProbedAt string `json:"probedAt,omitempty"`
ProbeErr string `json:"probeErr,omitempty"`
// CorsOK says the bucket answered a browser preflight for the
// forge's origin at the last probe. False does not refuse the
// binding — the multipart form uploads through the forge — but the
// page's scripted upload straight to the bucket needs the steward's
// CORS rule first, and the settings page shows it.
CorsOK bool `json:"corsOk"`
Inherited string `json:"inherited,omitempty"`
// Archives and DeleteObjects are the Phase 3 switches
// (bucket.policy): may uploads declare an archive type, and does a
// removal delete the object.
Archives bool `json:"archives"`
DeleteObjects bool `json:"deleteObjects"`
}
type bucketGetIn struct {
Repo string `json:"repo" validate:"required,max=64"`
}
type bucketSetIn struct {
Repo string `json:"repo" validate:"required,max=64"`
Endpoint string `json:"endpoint" validate:"required,max=512"`
Bucket string `json:"bucket" validate:"required,max=128"`
// Region is what the provider wants in the signature's scope;
// "auto" is R2's answer and what an empty one becomes.
Region string `json:"region" validate:"max=64"`
AccessKey string `json:"accessKey" validate:"required,max=256"`
// Secret may be empty on a repository that is already bound: the
// stored one is kept, so editing the public base or the limit does
// not mean retyping the key.
Secret string `json:"secret" validate:"max=512"`
PublicBase string `json:"publicBase" validate:"max=512"`
// Limit is bytes; zero takes the default of 100 MiB.
Limit int64 `json:"limit" validate:"min=0"`
}
type bucketProbeIn struct {
Repo string `json:"repo" validate:"required,max=64"`
}
type bucketUnsetIn struct {
Repo string `json:"repo" validate:"required,max=64"`
}
type bucketUnsetOut struct {
Repo string `json:"repo"`
Unbound bool `json:"unbound"`
}
// bucketInfo projects a row onto the answer, secret dropped.
func bucketInfo(repo string, b oltp.BucketBinding) BucketInfo {
return BucketInfo{Repo: repo, Bound: true, Endpoint: b.Endpoint, Bucket: b.Bucket, Region: b.Region,
AccessKey: b.AccessKey, PublicBase: b.PublicBase, Limit: b.LimitBytes,
ProbedAt: b.ProbedAt, ProbeErr: b.ProbeErr, CorsOK: b.CorsOK, Archives: b.Archives, DeleteObjects: b.DeleteObjects}
}
// BucketOrigins is the CSP question for a page under repo: which
// origins may it show bytes from and send bytes to. The bucket's
// endpoint (presigned URLs) and its public base (public downloads),
// as scheme+host; nothing for a repository with no binding, and a
// companion answers with its parent's. Read from the store rather
// than through bucket.get because the caller is a middleware that
// runs before any identity is resolved — the op's read gate would
// refuse it on every private repository, which is exactly where a
// steward keeps files worth a bucket.
func (d Deps) BucketOrigins(repo string) []string {
if d.Buckets == nil {
return nil
}
info, _, err := d.bucketOf(repo)
if err != nil || !info.Bound {
return nil
}
var out []string
for _, raw := range []string{info.Endpoint, info.PublicBase} {
if u, err := url.Parse(raw); err == nil && u.Scheme != "" && u.Host != "" {
if o := u.Scheme + "://" + u.Host; !slices.Contains(out, o) {
out = append(out, o)
}
}
}
return out
}
// bucketOf answers the binding uploads to repo would use: the
// repository's own, or its parent's when repo is a companion store.
// The second answer carries Inherited so a page can say so.
func (d Deps) bucketOf(repo string) (BucketInfo, oltp.BucketBinding, error) {
owner := repo
if parent := d.companionParent(repo); parent != "" {
owner = parent
}
b, ok, err := d.Buckets.Bucket(owner)
if err != nil {
return BucketInfo{}, oltp.BucketBinding{}, err
}
info := BucketInfo{Repo: repo}
if owner != repo {
info.Inherited = owner
}
if ok {
info = bucketInfo(repo, b)
info.Repo = repo
if owner != repo {
info.Inherited = owner
}
}
return info, b, nil
}
// checkEndpoint is the forge's rule for where it will send a
// steward's key: https, or plain http to a loopback host, which is
// where a test's or a developer's MinIO lives and nowhere a key
// travels a network in the clear.
func checkEndpoint(raw string) error {
u, err := url.Parse(raw)
if err != nil || u.Host == "" || u.Path != "" && u.Path != "/" || u.RawQuery != "" {
return errors.New("the endpoint must be a scheme and a host, with nothing after them")
}
if u.User != nil {
return errors.New("the endpoint must not carry a user or password; the key goes in its own fields")
}
switch u.Scheme {
case "https":
return nil
case "http":
host := u.Hostname()
if ip := net.ParseIP(host); (ip != nil && ip.IsLoopback()) || host == "localhost" {
return nil
}
return errors.New("the endpoint must be https; plain http is allowed only to a loopback host")
}
return errors.New("the endpoint must be an https url")
}
// probe runs the presigned PUT and DELETE and records what they said.
// The stamp is written whether or not the probe passed, so "last
// checked" stays answerable however long a binding has been broken.
func (d Deps) probe(ctx context.Context, repo string, b oltp.BucketBinding) string {
ctx, cancel := context.WithTimeout(ctx, probeTimeout)
defer cancel()
bucket := s3.Bucket{Endpoint: b.Endpoint, Name: b.Bucket, Region: b.Region, AccessKey: b.AccessKey, Secret: b.Secret}
msg := ""
if err := bucket.Probe(ctx, &http.Client{Timeout: probeTimeout}, time.Now()); err != nil {
msg = err.Error()
}
d.recordProbe(ctx, repo, bucket, msg)
return msg
}
// recordProbe stamps a probe's outcome beside the browser's half: a
// preflight from the forge's own origin, sent only once the forge's
// PUT passed, since a bucket that refuses the key has nothing to say
// about CORS yet. A forge with no --origin cannot ask on the
// browser's behalf and records the preflight as not passed, which
// the settings page words as such.
func (d Deps) recordProbe(ctx context.Context, repo string, bucket s3.Bucket, probeErr string) {
corsOK := false
if probeErr == "" && d.Origin != "" {
if err := bucket.Preflight(ctx, &http.Client{Timeout: probeTimeout}, d.Origin, time.Now()); err == nil {
corsOK = true
} else {
log.Printf("bucket: %s: %v", repo, err)
}
}
if err := d.Buckets.SetBucketProbe(repo, time.Now(), probeErr, corsOK); err != nil {
log.Printf("bucket: %s: recording probe: %v", repo, err)
}
}
// bucketSettingsEvent is the trail's line for a change here, worded
// with what changed and never the key.
func (d Deps) bucketSettingsEvent(ctx context.Context, repo, what string) {
if d.Events == nil {
return
}
if err := d.Events.Raise(evt.RepoSettings, repo, map[string]any{
"repo": repo, "actor": ops.User(ctx), "summary": what,
}); err != nil {
log.Printf("events: raise %s: %v", evt.RepoSettings, err)
}
}
// registerBuckets mounts bucket.get, bucket.set, bucket.probe and
// bucket.unset. Registered whether or not a store is wired, like the
// mirror ops, so the documented surface is the same on every forge
// and an unwired one answers unavailable.
func registerBuckets(r *ops.Registry, d Deps) error {
unwired := func() error {
if d.Buckets == nil {
return fmt.Errorf("this forge binds no buckets: %w", ops.ErrUnavailable)
}
return nil
}
if err := ops.Register(r, ops.Op[bucketGetIn, BucketInfo]{
Name: "bucket.get",
Caps: "a",
Doc: "The bucket this repository's uploads go to (RFC 0016): endpoint, bucket, region, access key, public base, limit and the last probe. `corsOk` says the bucket answered a browser preflight for the forge's origin at that probe; false means the page's scripted upload is blocked until the bucket's CORS rule allows the forge, while the multipart form still works. A companion store answers its parent's binding with `inherited` naming the parent. The secret is never returned.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/r/{repo}/bucket",
Run: func(ctx context.Context, in bucketGetIn) (BucketInfo, error) {
if err := unwired(); err != nil {
return BucketInfo{}, err
}
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return BucketInfo{}, err
}
info, _, err := d.bucketOf(in.Repo)
return info, err
},
}); err != nil {
return err
}
if err := ops.Register(r, ops.Op[bucketSetIn, BucketInfo]{
Name: "bucket.set",
Caps: "a",
Doc: "Bind (or re-bind) the bucket this repository's uploads go to. The endpoint must be https; region defaults to `auto` (R2); limit is bytes and defaults to 100 MiB. The binding is probed first — a presigned PUT and DELETE of one small object under `attachments/` — and refused with the probe's answer if either fails. An empty secret on a bound repository keeps the stored one. Refused on a mirror, which accepts no writes, and on a companion store, which uses its parent's bucket.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/r/{repo}/bucket",
Run: func(ctx context.Context, in bucketSetIn) (BucketInfo, error) {
if err := unwired(); err != nil {
return BucketInfo{}, err
}
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return BucketInfo{}, err
}
if m, _ := d.posture(in.Repo); m.Tracking {
return BucketInfo{}, fmt.Errorf("%w: %s mirrors %s and accepts no writes, uploads included", ops.ErrForbidden, in.Repo, m.URL)
}
if parent := d.companionParent(in.Repo); parent != "" {
return BucketInfo{}, fmt.Errorf("%s holds %s's forum and uses %s's bucket; bind it there: %w", in.Repo, parent, parent, ops.ErrInvalid)
}
in.Endpoint = strings.TrimRight(strings.TrimSpace(in.Endpoint), "/")
if err := checkEndpoint(in.Endpoint); err != nil {
return BucketInfo{}, fmt.Errorf("%v: %w", err, ops.ErrInvalid)
}
if in.PublicBase != "" {
in.PublicBase = strings.TrimRight(strings.TrimSpace(in.PublicBase), "/")
if u, err := url.Parse(in.PublicBase); err != nil || u.Host == "" || (u.Scheme != "https" && u.Scheme != "http") {
return BucketInfo{}, fmt.Errorf("the public base must be an http(s) url: %w", ops.ErrInvalid)
}
}
if in.Region == "" {
in.Region = "auto"
}
have, bound, err := d.Buckets.Bucket(in.Repo)
if err != nil {
return BucketInfo{}, err
}
if in.Secret == "" {
if !bound {
return BucketInfo{}, fmt.Errorf("a first binding needs the secret: %w", ops.ErrInvalid)
}
in.Secret = have.Secret
}
next := oltp.BucketBinding{Endpoint: in.Endpoint, Bucket: in.Bucket, Region: in.Region,
AccessKey: in.AccessKey, Secret: in.Secret, PublicBase: in.PublicBase, LimitBytes: in.Limit}
if next.LimitBytes <= 0 {
next.LimitBytes = oltp.DefaultBucketLimit
}
// Probed BEFORE it is written. A binding that does not
// work is a form error, not a row that fails every upload.
probeCtx, cancel := context.WithTimeout(ctx, probeTimeout)
defer cancel()
bucket := s3.Bucket{Endpoint: next.Endpoint, Name: next.Bucket, Region: next.Region,
AccessKey: next.AccessKey, Secret: next.Secret}
if err := bucket.Probe(probeCtx, &http.Client{Timeout: probeTimeout}, time.Now()); err != nil {
return BucketInfo{}, fmt.Errorf("the bucket refused a probe (%v); check the endpoint, the bucket and the key's permissions: %w", err, ops.ErrInvalid)
}
if err := d.Buckets.BindBucket(in.Repo, next); err != nil {
return BucketInfo{}, fmt.Errorf("%v: %w", err, ops.ErrInvalid)
}
d.recordProbe(probeCtx, in.Repo, bucket, "")
what := fmt.Sprintf("attachments go to bucket %s at %s", next.Bucket, hostOf(next.Endpoint))
if bound {
what = fmt.Sprintf("attachments bucket changed: %s at %s", next.Bucket, hostOf(next.Endpoint))
if have.Endpoint == next.Endpoint && have.Bucket == next.Bucket && have.AccessKey == next.AccessKey {
what = "attachments bucket settings changed"
}
}
d.bucketSettingsEvent(ctx, in.Repo, what)
info, _, err := d.bucketOf(in.Repo)
return info, err
},
}); err != nil {
return err
}
if err := ops.Register(r, ops.Op[bucketProbeIn, BucketInfo]{
Name: "bucket.probe",
Caps: "a",
Doc: "Probe the bound bucket again — a presigned PUT and DELETE, then a browser preflight from the forge's origin — and record the outcome. The answer carries `probedAt` and, when it failed, `probeErr`; `corsOk` is the preflight's verdict. A failed probe is an answer, not an error.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/r/{repo}/bucket/probe",
Run: func(ctx context.Context, in bucketProbeIn) (BucketInfo, error) {
if err := unwired(); err != nil {
return BucketInfo{}, err
}
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return BucketInfo{}, err
}
info, b, err := d.bucketOf(in.Repo)
if err != nil {
return BucketInfo{}, err
}
if !info.Bound {
return BucketInfo{}, fmt.Errorf("%s has no bucket bound: %w", in.Repo, ops.ErrNotFound)
}
owner := in.Repo
if info.Inherited != "" {
owner = info.Inherited
}
d.probe(ctx, owner, b)
info, _, err = d.bucketOf(in.Repo)
return info, err
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[bucketUnsetIn, bucketUnsetOut]{
Name: "bucket.unset",
Caps: "a",
Doc: "Forget the repository's bucket binding. The objects stay in the steward's bucket and the stubs stay in the DAG; binding a bucket with the same objects makes them live again.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/r/{repo}/bucket/unset",
Run: func(ctx context.Context, in bucketUnsetIn) (bucketUnsetOut, error) {
if err := unwired(); err != nil {
return bucketUnsetOut{}, err
}
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return bucketUnsetOut{}, err
}
if parent := d.companionParent(in.Repo); parent != "" {
return bucketUnsetOut{}, fmt.Errorf("%s uses %s's bucket; unbind it there: %w", in.Repo, parent, ops.ErrInvalid)
}
_, bound, err := d.Buckets.Bucket(in.Repo)
if err != nil {
return bucketUnsetOut{}, err
}
if !bound {
return bucketUnsetOut{Repo: in.Repo}, nil
}
if err := d.Buckets.UnbindBucket(in.Repo); err != nil {
return bucketUnsetOut{}, err
}
d.bucketSettingsEvent(ctx, in.Repo, "attachments bucket unbound")
return bucketUnsetOut{Repo: in.Repo, Unbound: true}, nil
},
})
}
// hostOf is the host of a URL for a sentence, or the string itself
// when it does not parse.
func hostOf(raw string) string {
if u, err := url.Parse(raw); err == nil && u.Host != "" {
return u.Host
}
return raw
}