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/delivery"
	"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 reads the placement seat's state of who holds which seat, every key; a test replaces it.
	seats func() (map[string][]byte, error)

	mu      sync.Mutex
	keys    map[string][]byte // seat-holders, as read
	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() (map[string][]byte, error) {
	state := stdio.State(delivery.SeatHoldersRead)
	names, err := state.Keys()
	if err != nil {
		return nil, err
	}
	keys := map[string][]byte{}
	for _, k := range names {
		e, err := state.Get(k)
		if err != nil {
			return nil, err
		}
		if e != nil {
			keys[k] = e.Value
		}
	}
	return keys, nil
}

// 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 placement seat's holders 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 placement seat's state of who holds which seat (novox/hq ADR
// 0571 clause 4), 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 {
		keys, err := s.seats()
		if err != nil {
			return false, err
		}
		s.keys, s.readAt, s.haveSet = keys, s.now(), true
	}
	holders, err := delivery.HoldersOf(s.keys, seat, node)
	if err != nil {
		return false, err
	}
	for _, h := range holders {
		if h.Node == node && h.Module == module {
			return true, nil
		}
	}
	return false, nil
}