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

307 lines
9.8 KiB
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
}