Serve object-store.link: a signed link to one object, signed with the bucket's link key, for the consumer and its shares; retire the root-signed minio_presigned_url (issue 1516)
This commit is contained in:
@@ -0,0 +1,306 @@
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user