Fetch a declaration's files given by digest from the bus's object bucket, verify them and report file-references (issue 1695, ADR 0486)

mesh-sdk is pinned to fix/1694-file-references 6c857e1c025f until it merges.

internal/apply/references.go

@@ -0,0 +1,95 @@
+package apply
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"unicode/utf8"
+
+	"encoding/base64"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+)
+
+// FetchObject reads the object of the declaration-files bucket named by name, no larger than limit bytes (novox/hq ADR
+// 0486 §1, §2). Nil is a node-engine with no link to the bus: a file by reference is then only the file already there.
+type FetchObject func(ctx context.Context, name string, limit int64) ([]byte, error)
+
+// ResolveFileReferences puts the content of every file given by digest and size into the declaration, so what is
+// applied afterwards is the declaration as it was before references existed (novox/hq ADR 0486 §1, issue 1695).
+//
+// **Nothing of the declaration is applied before every reference is resolved**: a missing object, one whose size or
+// sha256 differs from what the signed declaration names, or a fetch that fails, refuses the declaration whole. The
+// content is checked with File.Verify before it is kept, and is only ever written by the apply that follows.
+//
+// A file already on the machine whose content verifies is used as it is and fetches nothing: the reconcile of an
+// unchanged declaration, and a node that is asleep and has no bus, apply as they always did. One object is fetched
+// once however many files name its digest.
+func ResolveFileReferences(ctx context.Context, d *declaration.Declaration, fetch FetchObject) error {
+	fetched := map[string][]byte{}
+	for i, r := range d.Resources {
+		f, ok := r.(*declaration.File)
+		if !ok || !f.IsReference() {
+			continue
+		}
+		body, err := referencedContent(ctx, f, fetch, fetched)
+		if err != nil {
+			return err
+		}
+		inline := *f
+		inline.Digest, inline.Size = "", 0
+		if utf8.Valid(body) {
+			inline.Content = string(body)
+		} else {
+			inline.Bytes = base64.StdEncoding.EncodeToString(body)
+		}
+		d.Resources[i] = &inline
+	}
+	return nil
+}
+
+func referencedContent(ctx context.Context, f *declaration.File, fetch FetchObject,
+	fetched map[string][]byte) ([]byte, error) {
+	if body, ok := fetched[f.Digest]; ok {
+		return body, nil
+	}
+	if body, ok := alreadyThere(f); ok {
+		fetched[f.Digest] = body
+		return body, nil
+	}
+	name, err := declaration.ObjectName(f.Digest)
+	if err != nil {
+		return nil, fmt.Errorf("file %s (%s): %w", f.ID, f.Path, err)
+	}
+	if fetch == nil {
+		return nil, fmt.Errorf("file %s (%s) is given by digest %s and is not on this machine, and this node-engine "+
+			"has no link to the bus to fetch it from", f.ID, f.Path, f.Digest)
+	}
+	body, err := fetch(ctx, name, f.Size)
+	if err != nil {
+		return nil, fmt.Errorf("file %s (%s): fetching the object %s of the bucket %s: %w",
+			f.ID, f.Path, name, declaration.FileBucket, err)
+	}
+	if err := f.Verify(body); err != nil {
+		return nil, fmt.Errorf("%s (%s): %w; nothing was applied", f.ID, f.Path, err)
+	}
+	fetched[f.Digest] = body
+	return body, nil
+}
+
+// alreadyThere is the content of the file on the machine when it is what the reference names.
+func alreadyThere(f *declaration.File) ([]byte, bool) {
+	if !filepath.IsAbs(f.Path) {
+		return nil, false
+	}
+	info, err := os.Lstat(f.Path)
+	if err != nil || !info.Mode().IsRegular() || info.Size() != f.Size {
+		return nil, false
+	}
+	body, err := os.ReadFile(f.Path)
+	if err != nil || f.Verify(body) != nil {
+		return nil, false
+	}
+	return body, true
+}

internal/apply/references_test.go

@@ -0,0 +1,119 @@
+package apply
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+)
+
+func referenceTo(t *testing.T, path string, body []byte) *declaration.File {
+	t.Helper()
+	return &declaration.File{ID: "m.big", Type: declaration.TypeFile, Path: path,
+		Digest: declaration.DigestOf(body), Size: int64(len(body))}
+}
+
+func objects(m map[string][]byte, asked *[]string) FetchObject {
+	return func(_ context.Context, name string, limit int64) ([]byte, error) {
+		*asked = append(*asked, name)
+		b, ok := m[name]
+		if !ok {
+			return nil, errors.New("object not found")
+		}
+		return b, nil
+	}
+}
+
+func objectOf(body []byte) string {
+	name, _ := declaration.ObjectName(declaration.DigestOf(body))
+	return name
+}
+
+// 1695/AC1: both forms apply: a file by reference becomes the inline file, an inline one is left as it is.
+func TestAFileByReferenceBecomesTheInlineFile(t *testing.T) {
+	body := []byte(strings.Repeat("a line\n", 20000))
+	inline := &declaration.File{ID: "m.small", Type: declaration.TypeFile, Path: "/etc/small", Content: "x"}
+	d := &declaration.Declaration{Resources: []declaration.Resource{inline, referenceTo(t, "/nowhere/big", body)}}
+	var asked []string
+	if err := ResolveFileReferences(context.Background(), d, objects(map[string][]byte{objectOf(body): body}, &asked)); err != nil {
+		t.Fatal(err)
+	}
+	got := d.Resources[1].(*declaration.File)
+	if got.IsReference() || got.Content != string(body) || got.Path != "/nowhere/big" || d.Resources[0] != declaration.Resource(inline) {
+		t.Errorf("1695/AC1: resolved to %+v", got)
+	}
+	if len(asked) != 1 || asked[0] != objectOf(body) {
+		t.Errorf("1695/AC1: asked %v", asked)
+	}
+}
+
+// 1695/AC2: an object that differs by one byte, or in size, fails the whole resolve and gives no content.
+func TestAnObjectThatDiffersFailsTheApply(t *testing.T) {
+	body := []byte(strings.Repeat("x", 70000))
+	bad := append([]byte(nil), body...)
+	bad[3] ^= 1
+	for name, served := range map[string][]byte{"one byte": bad, "shorter": body[1:]} {
+		d := &declaration.Declaration{Resources: []declaration.Resource{referenceTo(t, "/nowhere/big", body)}}
+		var asked []string
+		err := ResolveFileReferences(context.Background(), d, objects(map[string][]byte{objectOf(body): served}, &asked))
+		if err == nil || !strings.Contains(err.Error(), "nothing was applied") {
+			t.Errorf("1695/AC2: %s: %v", name, err)
+		}
+		if !d.Resources[0].(*declaration.File).IsReference() {
+			t.Errorf("1695/AC2: %s: the reference was replaced by unverified content", name)
+		}
+	}
+}
+
+// 1695/AC3: a missing object, or a node with no bus, fails the apply naming the file.
+func TestAMissingObjectFailsTheApply(t *testing.T) {
+	body := []byte(strings.Repeat("y", 70000))
+	d := &declaration.Declaration{Resources: []declaration.Resource{referenceTo(t, "/nowhere/big", body)}}
+	var asked []string
+	if err := ResolveFileReferences(context.Background(), d, objects(nil, &asked)); err == nil ||
+		!strings.Contains(err.Error(), "m.big") || !strings.Contains(err.Error(), "declaration-files") {
+		t.Errorf("1695/AC3: missing object: %v", err)
+	}
+	if err := ResolveFileReferences(context.Background(), d, nil); err == nil || !strings.Contains(err.Error(), "no link to the bus") {
+		t.Errorf("1695/AC3: no bus: %v", err)
+	}
+}
+
+// 1695/AC4: a file already on the machine that verifies is used and nothing is fetched; one that does not is fetched.
+func TestAVerifiedFileOnTheMachineFetchesNothing(t *testing.T) {
+	body := []byte(strings.Repeat("z", 70000))
+	path := filepath.Join(t.TempDir(), "big")
+	if err := os.WriteFile(path, body, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	var asked []string
+	d := &declaration.Declaration{Resources: []declaration.Resource{referenceTo(t, path, body)}}
+	if err := ResolveFileReferences(context.Background(), d, objects(nil, &asked)); err != nil || len(asked) != 0 {
+		t.Errorf("1695/AC4: err %v, asked %v", err, asked)
+	}
+	if err := os.WriteFile(path, body[:100], 0o644); err != nil {
+		t.Fatal(err)
+	}
+	d = &declaration.Declaration{Resources: []declaration.Resource{referenceTo(t, path, body)}}
+	if err := ResolveFileReferences(context.Background(), d, objects(map[string][]byte{objectOf(body): body}, &asked)); err != nil || len(asked) != 1 {
+		t.Errorf("1695/AC4: a differing file: err %v, asked %v", err, asked)
+	}
+}
+
+// 1695/AC5: binary content comes back as bytes, and two files naming one digest fetch it once.
+func TestBinaryContentAndOneFetchPerDigest(t *testing.T) {
+	body := append([]byte{0xff, 0xfe, 0x00}, []byte(strings.Repeat("b", 70000))...)
+	d := &declaration.Declaration{Resources: []declaration.Resource{referenceTo(t, "/n/a", body), referenceTo(t, "/n/b", body)}}
+	var asked []string
+	if err := ResolveFileReferences(context.Background(), d, objects(map[string][]byte{objectOf(body): body}, &asked)); err != nil {
+		t.Fatal(err)
+	}
+	a, _ := d.Resources[0].(*declaration.File).Body()
+	if d.Resources[0].(*declaration.File).Bytes == "" || string(a) != string(body) || len(asked) != 1 {
+		t.Errorf("1695/AC5: bytes %q asked %v", d.Resources[0].(*declaration.File).Bytes[:4], asked)
+	}
+}

internal/link/files.go

@@ -0,0 +1,83 @@
+package link
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+	"github.com/nats-io/nats.go"
+)
+
+// FilesBus is a link that can read an object of the bucket the controller keeps a declaration's large files in
+// (novox/hq ADR 0486 §1, §2). Separate from Bus, as HaveBus is: a link that cannot is still a link, and a declaration
+// whose files are given by digest is then refused whole, naming that.
+type FilesBus interface {
+	// FileObject is the object of that name, no larger than limit bytes, read within ctx. **Reads by name only**: a
+	// node-engine's account may neither list the bucket nor write it (ADR 0486 §2).
+	FileObject(ctx context.Context, name string, limit int64) ([]byte, error)
+}
+
+type filesKey struct{}
+
+// WithFiles is ctx carrying the link a declaration's file references are fetched on.
+func WithFiles(ctx context.Context, bus any) context.Context {
+	if f, ok := bus.(FilesBus); ok {
+		return context.WithValue(ctx, filesKey{}, f)
+	}
+	return ctx
+}
+
+// FilesFrom is the fetch the apply's context carries, or nil when the node has no such link.
+func FilesFrom(ctx context.Context) func(context.Context, string, int64) ([]byte, error) {
+	if f, ok := ctx.Value(filesKey{}).(FilesBus); ok {
+		return f.FileObject
+	}
+	return nil
+}
+
+// FileObject reads one object by name from the declaration-files bucket. **Only binds**: the bucket is the controller's
+// to make, so one that is not there is said as what it is. Nothing waits longer than ctx allows.
+func (b OverNATS) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
+	js := b.JS
+	wait := 10 * time.Second
+	if at, ok := ctx.Deadline(); ok {
+		if wait = time.Until(at); wait <= 0 {
+			return nil, context.DeadlineExceeded
+		}
+	}
+	if b.Conn != nil {
+		bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
+		if err != nil {
+			return nil, err
+		}
+		js = bounded
+	}
+	store, err := js.ObjectStore(declaration.FileBucket)
+	if err != nil {
+		return nil, fmt.Errorf("the object bucket %s: %w", declaration.FileBucket, err)
+	}
+	res, err := store.Get(name, nats.Context(ctx))
+	if err != nil {
+		return nil, err
+	}
+	defer res.Close()
+	body, err := io.ReadAll(io.LimitReader(res, limit+1))
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(body)) > limit {
+		return nil, fmt.Errorf("the object is larger than the %d bytes the declaration names", limit)
+	}
+	return body, nil
+}
+
+func (l *natsLink) FileObject(ctx context.Context, name string, limit int64) ([]byte, error) {
+	return OverNATS{Conn: l.conn, JS: l.js}.FileObject(ctx, name, limit)
+}
+
+var (
+	_ FilesBus = OverNATS{}
+	_ FilesBus = (*natsLink)(nil)
+)

internal/link/files_nats_integration_test.go

@@ -0,0 +1,95 @@
+//go:build integration
+
+package link
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+	"github.com/nats-io/nats.go"
+)
+
+// 1695/AC7: against a real bus with the bucket declaration-files, over the bus's default maximum message size, a
+// node-engine's link reads three objects that together are over 3 MiB by name, verifies each with File.Verify, and
+// refuses an object that is absent, one larger than the declaration says, and one that differs by a byte.
+//
+//	MESH_TEST_NATS=nats://127.0.0.1:4222 go test -tags integration ./internal/link/ -run TestFilesAreFetchedByDigestFromTheBucket
+//
+// The bucket is made here: in the mesh the controller makes and fills it, and the node-engine only reads.
+func TestFilesAreFetchedByDigestFromTheBucket(t *testing.T) {
+	conn, js := aBus(t)
+	store, err := js.ObjectStore(declaration.FileBucket)
+	if err != nil {
+		if store, err = js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: declaration.FileBucket}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var files []*declaration.File
+	bodies := map[string][]byte{}
+	total := 0
+	for i, b := range []byte{'a', 'b', 'c'} {
+		body := append(bytes.Repeat([]byte{b}, 1100*1024), byte(i))
+		total += len(body)
+		f := &declaration.File{ID: "m.f" + string(b), Type: declaration.TypeFile, Path: "/x/" + string(b),
+			Digest: declaration.DigestOf(body), Size: int64(len(body))}
+		name, err := declaration.ObjectName(f.Digest)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := store.PutBytes(name, body); err != nil {
+			t.Fatal(err)
+		}
+		files, bodies[f.ID] = append(files, f), body
+	}
+	if total <= 3<<20 {
+		t.Fatalf("the objects add up to %d bytes, not over 3 MiB", total)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	bus := OverNATS{Conn: conn, JS: js}
+	fetch := FilesFrom(WithFiles(ctx, bus))
+	if fetch == nil {
+		t.Fatal("a link over the bus carries no fetch")
+	}
+	for _, f := range files {
+		name, _ := declaration.ObjectName(f.Digest)
+		got, err := fetch(ctx, name, f.Size)
+		if err != nil {
+			t.Fatalf("fetching %s: %v", f.ID, err)
+		}
+		if err := f.Verify(got); err != nil {
+			t.Errorf("%s fetched and does not verify: %v", f.ID, err)
+		}
+		if !bytes.Equal(got, bodies[f.ID]) {
+			t.Errorf("%s fetched with other content", f.ID)
+		}
+	}
+
+	name, _ := declaration.ObjectName(files[0].Digest)
+	if _, err := fetch(ctx, name, files[0].Size-1); err == nil || !strings.Contains(err.Error(), "larger") {
+		t.Errorf("an object larger than the declaration says was read: %v", err)
+	}
+	absent := strings.Repeat("0", 64)
+	if _, err := fetch(ctx, absent, 10); err == nil {
+		t.Error("an absent object was read")
+	}
+	// A changed byte is stored under the digest of the original: what the controller never does and an attacker might.
+	bad := append([]byte(nil), bodies["m.fa"]...)
+	bad[5] ^= 1
+	swapped := strings.Repeat("1", 64)
+	if _, err := store.PutBytes(swapped, bad); err != nil {
+		t.Fatal(err)
+	}
+	got, err := fetch(ctx, swapped, files[0].Size)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := files[0].Verify(got); err == nil {
+		t.Error("an object that differs by a byte verified")
+	}
+}

internal/link/queue.go

@@ -214,7 +214,7 @@ func (q *Queue) step(ctx context.Context) bool {
 			Superseded: declaredIn(latest.Body())}, setAside, old)
 	}
 
-	report := handleBody(ctx, q.Membership, latest.Body(), q.Apply)
+	report := handleBody(q.withFiles(ctx), q.Membership, latest.Body(), q.Apply)
 	switch {
 	case report.OlderThan != nil:
 		// **Refused, counted and said** (rule 2): one line here, the count on this and every later
@@ -260,7 +260,7 @@ func (q *Queue) reconcile(ctx context.Context, asked bool) {
 		}
 		return
 	}
-	report, ok := q.Reconcile(ctx)
+	report, ok := q.Reconcile(q.withFiles(ctx))
 	if !ok {
 		if asked {
 			q.Say("asked to say what this node last applied, and the reconcile could not say it")
@@ -457,6 +457,18 @@ func (q *Queue) attach(ctx context.Context, bus Bus) {
 	q.mu.Unlock()
 }
 
+// withFiles is ctx carrying the link now open, so an apply can fetch a declaration's files given by digest (novox/hq ADR
+// 0486, issue 1695). A queue with no link open carries none.
+func (q *Queue) withFiles(ctx context.Context) context.Context {
+	q.mu.Lock()
+	bus := q.bus
+	q.mu.Unlock()
+	if bus == nil {
+		return ctx
+	}
+	return WithFiles(ctx, bus)
+}
+
 // detach is the link closing: it waits for a delivery in hand to be applied and said — an apply that
 // stood aside for its successor is reported on this link before it is let go (issue 264) — and then
 // nothing more is said on it. A reconcile in hand is not waited for: a link that was dropped or roused

internal/profile/detectors.go

@@ -29,6 +29,11 @@ const (
 	// an unprivileged user, never by a version string or a sysctl alone.
 	CapUserNamespaces = "user-namespaces"
 
+	// CapFileReferences says this node-engine reads a declaration's file given as a digest and size, fetches it from the
+	// bus's object bucket and checks it before writing (novox/hq ADR 0486 §5, issue 1695). A fact about the program, as
+	// CapContainerGroups: the controller sends references only to a node that reports it.
+	CapFileReferences = "file-references"
+
 	// The network manager this machine runs, one capability per dialect (novox/hq ADR 0161): the
 	// uplink seat's holder declares its own, so the holder for a manager the machine does not run
 	// is refused the way any missing capability is, naming it. Active, not installed — a machine
@@ -152,6 +157,19 @@ func (privileged) Detect(context.Context) Verdict {
 	}
 }
 
+// fileReferences reports that this node-engine reads a file given by digest. True of the program that is running.
+type fileReferences struct{}
+
+func (fileReferences) Name() string { return CapFileReferences }
+
+func (fileReferences) Detect(context.Context) Verdict {
+	return Verdict{
+		Name: CapFileReferences, Present: true,
+		Detail: "fetches a file given by digest from the bus's object bucket and verifies it before writing",
+		How:    "a fact about this program, not probed: the node-engine that answers reads the field",
+	}
+}
+
 // graphicalSession reports whether anything could display a window here.
 //
 // Environment rather than a probe, because a display server is reachable through a socket a
@@ -190,6 +208,7 @@ func Default(runner Runner) []Detector {
 	}
 	return []Detector{
 		privileged{},
+		fileReferences{},
 		graphicalSession{},
 		seat{},
 		battery{},

internal/profile/file_references_test.go

@@ -0,0 +1,18 @@
+package profile
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// 1695/AC6: the node's report carries the capability file-references.
+func TestTheProfileSaysTheNodeEngineReadsFileReferences(t *testing.T) {
+	got := Detect(context.Background(), Default(func(context.Context, string, ...string) (string, error) {
+		return "", nil
+	}), time.Second)
+	v := verdictOf(t, got, CapFileReferences)
+	if !v.Present || v.Detail == "" || v.How == "" || CapFileReferences != "file-references" {
+		t.Errorf("1695/AC6: file-references = %+v", v)
+	}
+}