mesh-controller/examples/objectstore-provisioner/main.go

// A provisioner for buckets, in the form the mesh expects one.
//
// The same contract as `examples/postgres-provisioner`, against a different kind of thing — and
// that is the point of it existing. The mesh generated a secret, sealed it to the machine that
// must accept it, and discarded the plaintext, so it cannot tell an object store to start
// accepting it. Something on that machine reads what the host wrote and makes it true.
//
// **It is an example, not part of the control plane** (novox/hq ADR 0001). A real one ships with
// the module that ships the object store. What lives here is the contract, written as something
// that runs so it can be read rather than described.
//
// What it is given, both written by the host from an ordinary declaration:
//
//	$GRANTS/mesh.json          every consumer, what it asked for, and where its credential is
//	$GRANTS/<node>.secret      one consumer's secret key, alone in the file
//
// **Why it drives the vendor's own command line rather than an SDK.** MinIO's admin API encrypts
// its request bodies, which is why a separate admin library exists; pulling that in would add a
// system-metrics dependency tree to a repository that has none, to create a user. A provisioner
// is allowed to know how to operate the thing it provisions — that is the whole of its job — and
// this way the example stays about the contract instead of about somebody's request signing.
package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"
)

// mark is what this provisioner names the access keys it owns.
//
// So it never removes one a person made by hand — the mesh's own rule about origins, one level
// down (novox/hq 04-ISSUES/010). A provisioner that deleted every user it did not recognise would
// be one nobody could safely run against a store that predates it.
const mark = "mesh_"

// alias is the name `mc` keeps its connection under. Local to this process's config directory.
const alias = "store"

// contribution is one consumer, as the mesh described it.
type contribution struct {
	From string `json:"from"`
	// Node is empty for a module on this machine, which is asking for something local and is not
	// this provisioner's business.
	Node   string         `json:"node"`
	Secret string         `json:"secret"`
	Values map[string]any `json:"values"`
}

type manifest struct {
	Requirement string         `json:"requirement"`
	Given       []contribution `json:"given"`
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	if len(os.Args) > 1 && os.Args[1] == "--watch" {
		if err := watch(ctx); err != nil {
			fmt.Fprintf(os.Stderr, "mesh-provision-objectstore: %v\n", err)
			os.Exit(1)
		}
		return
	}
	if err := run(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "mesh-provision-objectstore: %v\n", err)
		os.Exit(1)
	}
}

// watch reconciles now, and again whenever what the mesh delivered changes.
//
// By polling rather than watching the filesystem: the host writes atomically, so the file is
// replaced rather than modified, and an inotify watch on the path stops seeing anything after the
// first replacement — a watcher that silently stops working.
func watch(ctx context.Context) error {
	const every = 10 * time.Second
	var last string

	for {
		state, err := given()
		switch {
		case err != nil:
			// Said and retried. A provisioner that exits because the mesh has not written
			// anything yet has to be restarted by hand after the first push.
			fmt.Fprintf(os.Stderr, "cannot read what was granted: %v\n", err)
		case state != last:
			if err := run(ctx); err != nil {
				// Reported and retried. The usual reason is that the store has not finished
				// starting, and giving up would mean a module that works only when two
				// containers happen to come up in the right order.
				fmt.Fprintf(os.Stderr, "%v\n", err)
			} else {
				last = state
			}
		}

		select {
		case <-ctx.Done():
			return nil
		case <-time.After(every):
		}
	}
}

// grantsDir is where the host was told to write what this provisioner is given.
func grantsDir() string {
	if d := os.Getenv("GRANTS"); d != "" {
		return d
	}
	return "/var/lib/objectstore/grants"
}

// given is everything the mesh has delivered, as one string, so a change to any of it is one
// comparison.
//
// Credentials are included by their **digest**, never their content: this is compared, held in
// memory for the life of the process, and printed on failure, and a secret belongs in none of
// those when a hash answers the same question.
func given() (string, error) {
	entries, err := os.ReadDir(grantsDir())
	if err != nil {
		return "", err
	}
	var names []string
	for _, e := range entries {
		names = append(names, e.Name())
	}
	sort.Strings(names)

	sum := sha256.New()
	for _, name := range names {
		body, err := os.ReadFile(filepath.Join(grantsDir(), name))
		if err != nil {
			return "", err
		}
		fmt.Fprintf(sum, "%s:%x\n", name, sha256.Sum256(body))
	}
	return hex.EncodeToString(sum.Sum(nil)), nil
}

func run(ctx context.Context) error {
	raw, err := os.ReadFile(filepath.Join(grantsDir(), "mesh.json"))
	if err != nil {
		if os.IsNotExist(err) {
			// Nothing has been granted here. Not a failure: a provider with no consumers is an
			// ordinary state, and one this must be able to reach from any other.
			fmt.Printf("nothing has been granted to this machine\n")
			return nil
		}
		return err
	}
	var m manifest
	if err := json.Unmarshal(raw, &m); err != nil {
		return fmt.Errorf("the manifest at %s is not readable: %w", grantsDir(), err)
	}

	if err := connect(ctx); err != nil {
		return err
	}

	// **Reconciling, not applying a change.** It runs after every declaration and is never told
	// what changed, so it must reach the same state from wherever it starts.
	wanted := map[string]bool{}
	for _, c := range sorted(m.Given) {
		if c.Node == "" {
			continue
		}
		bucket, _ := c.Values["bucket"].(string)
		if bucket == "" {
			return fmt.Errorf("%s asked for a bucket and did not name it", c.Node)
		}
		if err := usableBucketName(bucket); err != nil {
			return fmt.Errorf("%s asked for a bucket named %q: %w", c.Node, bucket, err)
		}
		secret, err := os.ReadFile(c.Secret)
		if err != nil {
			// The manifest says there is a credential and the host has not written it. Refused
			// rather than creating a user with no secret — a login nothing can use, which
			// nothing would report until something tried to connect.
			return fmt.Errorf("%s's credential should be at %s and is not there", c.Node, c.Secret)
		}

		// One access key per consumer, and a consumer is a module on a machine (novox/hq
		// 04-ISSUES/022) — otherwise every service on a node shares one key, and the policy that
		// confines each to its own bucket confines none of them.
		key := mark + c.Node + "_" + c.From
		wanted[key] = true
		if err := ensureBucket(ctx, bucket); err != nil {
			return err
		}
		if err := ensureUser(ctx, key, strings.TrimSpace(string(secret))); err != nil {
			return err
		}
		if err := ensureOnlyThatBucket(ctx, key, bucket); err != nil {
			return err
		}
	}

	// And everything this provisioner made that nobody asks for any more. **The half usually
	// missing**: a consumer that goes away otherwise keeps a working key for ever, and nothing
	// says so.
	return revokeOrphans(ctx, wanted)
}

// sorted puts the consumers in a stable order, so two runs over the same input do the same things
// in the same sequence and a log can be compared against another.
func sorted(given []contribution) []contribution {
	out := append([]contribution(nil), given...)
	sort.Slice(out, func(i, j int) bool {
		if out[i].Node != out[j].Node {
			return out[i].Node < out[j].Node
		}
		return out[i].From < out[j].From
	})
	return out
}

// usableBucketName refuses a name the store would refuse, but says so in terms of the module that
// asked rather than in terms of a REST error nobody can trace back.
//
// Deliberately narrower than the store's own rule: this rejects what is ambiguous as well as what
// is invalid, because a bucket named `Photos` that arrives as `photos` is a module that works
// until somebody looks.
func usableBucketName(name string) error {
	if len(name) < 3 || len(name) > 63 {
		return fmt.Errorf("a bucket name is 3 to 63 characters")
	}
	for _, r := range name {
		if (r < 'a' || r > 'z') && (r < '0' || r > '9') && r != '-' {
			return fmt.Errorf("only lower-case letters, digits and dashes are usable, and %q is not", r)
		}
	}
	if name[0] == '-' || name[len(name)-1] == '-' {
		return fmt.Errorf("it may not begin or end with a dash")
	}
	return nil
}

// onlyThatBucket is the policy a consumer gets: its own bucket, and nothing else in the store.
//
// **Written per consumer rather than shared.** A single policy naming every bucket would grow a
// line each time somebody asks for one, and every existing consumer would silently gain access to
// the new one.
func onlyThatBucket(bucket string) string {
	return fmt.Sprintf(`{
  "Version": "2012-10-17",
  "Statement": [
    {"Effect": "Allow",
     "Action": ["s3:ListBucket", "s3:GetBucketLocation"],
     "Resource": ["arn:aws:s3:::%s"]},
    {"Effect": "Allow",
     "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
     "Resource": ["arn:aws:s3:::%s/*"]}
  ]
}`, bucket, bucket)
}

func mc(ctx context.Context, args ...string) (string, error) {
	// Its own configuration directory, so this never reads or writes whatever a person running
	// `mc` on this machine has set up.
	cmd := exec.CommandContext(ctx, "mc", append([]string{"--config-dir", "/tmp/mesh-mc"}, args...)...)
	out, err := cmd.CombinedOutput()
	if err != nil {
		// The command is named without its arguments: an argument here can be a secret.
		return string(out), fmt.Errorf("mc %s: %w\n%s", args[0], err, strings.TrimSpace(string(out)))
	}
	return string(out), nil
}

// connect points `mc` at the store, using the root credential this provisioner was given.
func connect(ctx context.Context) error {
	endpoint := os.Getenv("MESH_OBJECTSTORE_URL")
	if endpoint == "" {
		endpoint = "http://127.0.0.1:9000"
	}
	user := os.Getenv("MESH_OBJECTSTORE_ROOT_USER")
	file := os.Getenv("MESH_OBJECTSTORE_ROOT_PASSWORD_FILE")
	if user == "" || file == "" {
		// Named together, because a provisioner that starts with half its credential fails later
		// against the store and reads as the store being wrong.
		return fmt.Errorf("MESH_OBJECTSTORE_ROOT_USER and MESH_OBJECTSTORE_ROOT_PASSWORD_FILE " +
			"must both be set: this provisions the store and has to authenticate to it")
	}
	// From a file, never an environment variable: the environment of a process is readable by
	// anything that can see the process, and this one is the store's root.
	password, err := os.ReadFile(file)
	if err != nil {
		return fmt.Errorf("the store's root password should be at %s and is not there", file)
	}
	_, err = mc(ctx, "alias", "set", alias, endpoint, user, strings.TrimSpace(string(password)))
	return err
}

func ensureBucket(ctx context.Context, bucket string) error {
	_, err := mc(ctx, "mb", "--ignore-existing", alias+"/"+bucket)
	return err
}

// ensureUser sets the secret every time rather than only on creation.
//
// The mesh replaces the file when it rotates, and a provisioner that only ever created would
// leave the old key working — a rotation that reports success and changes nothing.
func ensureUser(ctx context.Context, key, secret string) error {
	_, err := mc(ctx, "admin", "user", "add", alias, key, secret)
	return err
}

func ensureOnlyThatBucket(ctx context.Context, key, bucket string) error {
	name := key + "-" + bucket
	path := filepath.Join("/tmp", name+".json")
	if err := os.WriteFile(path, []byte(onlyThatBucket(bucket)), 0o600); err != nil {
		return err
	}
	defer os.Remove(path)

	// Replaced rather than created-if-absent: the bucket a consumer asks for can change, and a
	// policy that was only ever created would keep granting the old one as well.
	_, _ = mc(ctx, "admin", "policy", "remove", alias, name)
	if _, err := mc(ctx, "admin", "policy", "create", alias, name, path); err != nil {
		return err
	}
	_, err := mc(ctx, "admin", "policy", "attach", alias, name, "--user", key)
	if err != nil && strings.Contains(err.Error(), "already") {
		// Attaching a policy that is already attached is the state being asked for.
		return nil
	}
	return err
}

// revokeOrphans removes the keys this provisioner made that nobody asks for any more.
func revokeOrphans(ctx context.Context, wanted map[string]bool) error {
	out, err := mc(ctx, "admin", "user", "list", alias, "--json")
	if err != nil {
		return err
	}
	for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
		if line == "" {
			continue
		}
		var u struct {
			AccessKey string `json:"accessKey"`
		}
		if json.Unmarshal([]byte(line), &u) != nil {
			continue
		}
		if !strings.HasPrefix(u.AccessKey, mark) || wanted[u.AccessKey] {
			continue
		}
		if _, err := mc(ctx, "admin", "user", "remove", alias, u.AccessKey); err != nil {
			return err
		}
		fmt.Printf("revoked %s, which nothing asks for any more\n", u.AccessKey)
	}
	return nil
}