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

291 lines
9.1 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/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
}