mesh-catalog/modules/minio/cmd/object-store/server.go

package main

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/objectlink"
)

// DefaultSeconds is the life of a link when none is asked; MaxSeconds is the longest (ADR 0468 §3).
const (
	DefaultSeconds = 300
	MaxSeconds     = 600
	holdersFor     = 60 * time.Second
)

// storeConfig is the store as this machine reaches it. The file is rendered by the mesh from files/object-store.json
// with the same facts the s3-bucket binding renders for a consumer (`at` is the providing machine's, `port` the one
// it gave the S3 endpoint), so a link names the host and port every consumer's own file names.
type storeConfig struct {
	Scheme string `json:"scheme"`
	At     string `json:"at"`
	Port   string `json:"port"`
	Region string `json:"region"`
}

func (c storeConfig) endpoint() string { return c.Scheme + "://" + c.At + ":" + c.Port }

func readConfig(path string) (storeConfig, error) {
	var c storeConfig
	raw, err := os.ReadFile(path)
	if err != nil {
		return c, fmt.Errorf("the store's address is unreadable (%s): %w", path, err)
	}
	if err := json.Unmarshal(raw, &c); err != nil {
		return c, fmt.Errorf("the store's address file %s is not JSON: %w", path, err)
	}
	if c.Scheme == "" || c.At == "" || c.Port == "" {
		return c, fmt.Errorf("the store's address file %s lacks a scheme, an at or a port", path)
	}
	return c, nil
}

// linkKeyID is the access key id of a bucket's link key, as the provisioner makes it (modules/minio/links.ts).
func linkKeyID(bucket string) string {
	sum := sha256.Sum256([]byte(bucket))
	return "link" + hex.EncodeToString(sum[:])[:16]
}

type server struct {
	configPath, linksDir, receives string
	now                            func() time.Time
	// seats asks the controller for its seats; a test replaces it.
	seats func() (json.RawMessage, error)

	mu      sync.Mutex
	holders map[string]bool // "<seat>/<node>.<module>"
	readAt  time.Time
	haveSet bool
}

func newServer(config, links, receives string) (*server, error) {
	for word, v := range map[string]string{ConfigWord: config, LinksWord: links, ReceivesWord: receives} {
		if strings.TrimSpace(v) == "" {
			return nil, fmt.Errorf("%s is not given: the manifest names it", word)
		}
	}
	return &server{configPath: config, linksDir: links, receives: receives, now: time.Now, seats: askSeats}, nil
}

func askSeats() (json.RawMessage, error) {
	return stdio.Ask("seat:mesh-controller.seats", map[string]any{})
}

// link answers one ask.
func (s *server) link(caller string, args map[string]any) (any, error) {
	bucket, _ := args["bucket"].(string)
	key, _ := args["key"].(string)
	method, _ := args["method"].(string)
	if bucket == "" {
		return nil, errors.New("link: bucket is needed")
	}
	if key == "" {
		return nil, errors.New("link: key is needed")
	}
	if method != "get" && method != "put" {
		return nil, fmt.Errorf("link: method %q is not get or put", method)
	}
	seconds := DefaultSeconds
	if v, given := args["seconds"]; given {
		f, ok := v.(float64)
		if !ok || f != float64(int(f)) {
			return nil, fmt.Errorf("link: seconds %v is not a whole number", v)
		}
		seconds = int(f)
	}
	if seconds < 1 || seconds > MaxSeconds {
		return nil, fmt.Errorf("link: seconds %d is outside 1 to %d: a link lives at most ten minutes (ADR 0468 §3)", seconds, MaxSeconds)
	}
	node, module, ok := splitCaller(caller)
	if !ok {
		return nil, fmt.Errorf("link: the caller is %q: the runtime did not prove who asks, and a link is not given to an unproven caller", caller)
	}
	grant, err := s.grantOf(bucket)
	if err != nil {
		return nil, err
	}
	if err := s.allowed(grant, node, module, method); err != nil {
		return nil, fmt.Errorf("link: %s.%s may not %s in bucket %s: %w", node, module, method, bucket, err)
	}
	cfg, err := readConfig(s.configPath)
	if err != nil {
		return nil, err
	}
	secret, err := os.ReadFile(filepath.Join(s.linksDir, bucket+".secret"))
	if err != nil || strings.TrimSpace(string(secret)) == "" {
		return nil, fmt.Errorf("link: bucket %s has no link key secret yet (%s); minio's provisioner makes it", bucket, filepath.Join(s.linksDir, bucket+".secret"))
	}
	now := s.now()
	url, err := objectlink.Presign(objectlink.Request{
		Method: strings.ToUpper(method), Endpoint: cfg.endpoint(), Bucket: bucket, Key: key,
		AccessKey: linkKeyID(bucket), SecretKey: strings.TrimSpace(string(secret)), Region: cfg.Region,
		Expires: time.Duration(seconds) * time.Second, Now: now,
	})
	if err != nil {
		return nil, err
	}
	return map[string]any{"link": url, "method": method, "seconds": seconds, "expires-at": now.Add(time.Duration(seconds) * time.Second).UTC().Format(time.RFC3339)}, nil
}

// splitCaller splits the bus user `<node>.<module>` the runtime proved.
func splitCaller(caller string) (node, module string, ok bool) {
	if caller == "" || caller == stdio.Unproven {
		return "", "", false
	}
	node, module, ok = strings.Cut(caller, ".")
	if !ok || node == "" || module == "" || strings.Contains(module, ".") {
		return "", "", false
	}
	return node, module, true
}

// grant is what the mesh delivered for one bucket: its consumer and the consumer's shares.
type grant struct {
	consumerModule, consumerNode string
	shares                       []share
}

type share struct {
	To      string   `json:"to"`
	Methods []string `json:"methods"`
}

// grantOf reads the grant file now (never cached, so a change of shares applies to the next ask) and finds the entry
// whose derived bucket is this one. The entry's shape is the mesh's Contribution (mesh-sdk go/catalogue/declaration.go
// `Contribution`: from, node, as, values, derived) inside `given` (receivedFile in the same file).
func (s *server) grantOf(bucket string) (grant, error) {
	raw, err := os.ReadFile(s.receives)
	if err != nil {
		return grant{}, fmt.Errorf("link: the grant file is unreadable (%s): %w", s.receives, err)
	}
	var doc struct {
		Given []struct {
			From   string `json:"from"`
			Node   string `json:"node"`
			Values struct {
				Shares []share `json:"shares"`
			} `json:"values"`
			Derived struct {
				Bucket string `json:"bucket"`
			} `json:"derived"`
		} `json:"given"`
	}
	if err := json.Unmarshal(raw, &doc); err != nil {
		return grant{}, fmt.Errorf("link: the grant file %s is not JSON: %w", s.receives, err)
	}
	for _, g := range doc.Given {
		if g.Derived.Bucket == bucket {
			return grant{consumerModule: g.From, consumerNode: g.Node, shares: g.Values.Shares}, nil
		}
	}
	return grant{}, fmt.Errorf("link: no grant names bucket %s: nobody is given it", bucket)
}

// allowed judges the caller: the bucket's consumer, or a share for this method naming the caller's module, or a seat
// the caller's module holds.
func (s *server) allowed(g grant, node, module, method string) error {
	if g.consumerModule == module && (g.consumerNode == "" || g.consumerNode == node) {
		return nil
	}
	// A share names the caller when its `to` is the caller's module, or a seat the caller's module holds. The
	// controller is asked only for a seat share, and only when nothing earlier decided.
	names := func(sh share) (bool, error) {
		if sh.To == module {
			return true, nil
		}
		if seat, isSeat := strings.CutPrefix(sh.To, "seat:"); isSeat {
			held, err := s.holds(seat, node, module)
			if err != nil {
				return false, fmt.Errorf("a share names seat %s and the controller's seats cannot be read to tell whether this caller holds it: %w", seat, err)
			}
			return held, nil
		}
		return false, nil
	}
	var unread error
	for _, sh := range g.shares {
		if !has(sh.Methods, method) {
			continue
		}
		named, err := names(sh)
		if err != nil {
			unread = err
			continue
		}
		if named {
			return nil
		}
	}
	if unread != nil {
		return unread
	}
	for _, sh := range g.shares {
		if has(sh.Methods, method) {
			continue
		}
		if named, err := names(sh); err == nil && named {
			return fmt.Errorf("the bucket's shares give this caller %s, not %s", strings.Join(sh.Methods, " and "), method)
		}
	}
	return errors.New("it is not the bucket's consumer and no share of the bucket names it for that method")
}

func has(list []string, v string) bool {
	for _, x := range list {
		if x == v {
			return true
		}
	}
	return false
}

// holds says whether node.module holds the seat, from the controller's seats, read at most once a minute. A read that
// failed is not cached: the next ask reads again.
func (s *server) holds(seat, node, module string) (bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.haveSet || s.now().Sub(s.readAt) >= holdersFor {
		raw, err := s.seats()
		if err != nil {
			return false, err
		}
		set, err := holdersIn(raw)
		if err != nil {
			return false, err
		}
		s.holders, s.readAt, s.haveSet = set, s.now(), true
	}
	return s.holders[seat+"/"+node+"."+module], nil
}

// holdersIn reads the controller's `seats` answer (answer.seats[].holders[] of {node, module}, and answer.outside[] of
// {Claim, Node, Module}) into "<seat>/<node>.<module>". The answer may arrive wrapped in {"answer": ...}.
func holdersIn(raw json.RawMessage) (map[string]bool, error) {
	var doc struct {
		Answer json.RawMessage `json:"answer"`
		Seats  []struct {
			Seat    string `json:"seat"`
			Holders []struct {
				Node   string `json:"node"`
				Module string `json:"module"`
			} `json:"holders"`
		} `json:"seats"`
		Outside []struct {
			Claim, Node, Module string
		} `json:"outside"`
	}
	if err := json.Unmarshal(raw, &doc); err != nil {
		return nil, fmt.Errorf("the controller's seats answer is not JSON: %w", err)
	}
	if len(doc.Answer) > 0 && string(doc.Answer) != "null" {
		return holdersIn(doc.Answer)
	}
	if doc.Seats == nil && doc.Outside == nil {
		return nil, errors.New("the controller's seats answer holds neither seats nor outside claims")
	}
	out := map[string]bool{}
	for _, s := range doc.Seats {
		for _, h := range s.Holders {
			out[s.Seat+"/"+h.Node+"."+h.Module] = true
		}
	}
	for _, o := range doc.Outside {
		out[o.Claim+"/"+o.Node+"."+o.Module] = true
	}
	return out, nil
}