mesh-controller/internal/broker/declaration_files_live_test.go

//go:build integration

package broker

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/bcrypt"
)

// nodeEnginesFileObject is the node-engine's own call (mesh-host internal/link files.go, OverNATS.FileObject), as the
// engine makes it: the object's metadata by a direct get of one subject, then its chunks through an ordered consumer
// bound to the stream. It is copied here so the grants are judged by the call that has to work, and its subjects are the
// ones the grant names (issue 1697, notes 4 and 5); the node-engine's own integration test holds the other copy.
func nodeEnginesFileObject(ctx context.Context, nc *nats.Conn, name string) ([]byte, error) {
	js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
	if err != nil {
		return nil, err
	}
	stream := "OBJ_" + DeclarationFilesBucket
	meta, err := js.GetLastMsg(stream, "$O."+DeclarationFilesBucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)),
		nats.DirectGet(), nats.Context(ctx))
	if err != nil {
		return nil, err
	}
	var info struct {
		NUID string `json:"nuid"`
		Size uint64 `json:"size"`
	}
	if err := json.Unmarshal(meta.Data, &info); err != nil {
		return nil, err
	}
	sub, err := js.SubscribeSync("$O."+DeclarationFilesBucket+".C."+info.NUID, nats.OrderedConsumer(), nats.BindStream(stream))
	if err != nil {
		return nil, err
	}
	defer sub.Unsubscribe()
	body := make([]byte, 0, info.Size)
	for uint64(len(body)) < info.Size {
		msg, err := sub.NextMsgWithContext(ctx)
		if err != nil {
			return nil, err
		}
		body = append(body, msg.Data...)
	}
	return body, nil
}

// **The bucket's grants on a real server, judged by the node-engine's own call** (novox/hq ADR 0486 §2, issue 1697,
// 1697/AC2): the controller's account writes an object, a node-engine's account reads it by its name, a large object
// included, and is refused a list of the bucket (a stream info asking for every subject), a message get by sequence, a
// direct get by sequence, a consumer on the names, and a write; another module's account is refused all of them.
func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *testing.T) { // 1697/AC2
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	users, err := Users(Records{Nodes: []string{"alpha"}, Assigned: map[string][]Declared{
		"alpha": {{Module: RuntimeModule}, {Module: "other-module"}},
	}})
	if err != nil {
		t.Fatal(err)
	}
	for i := range users {
		users[i].PasswordHash = string(hash)
	}
	accounts, err := ComposeAccounts(users)
	if err != nil {
		t.Fatal(err)
	}
	conf := filepath.Join(t.TempDir(), "bus.conf")
	if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
		`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
		t.Fatal(err)
	}
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatalf("the composed accounts do not parse: %v", err)
	}
	opts.NoLog, opts.NoSigs = true, true
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(s.Shutdown)

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	connect := func(user string) (*nats.Conn, chan error) {
		t.Helper()
		refused := make(chan error, 16)
		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
			nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
		if err != nil {
			t.Fatalf("%s cannot connect: %v", user, err)
		}
		t.Cleanup(nc.Close)
		return nc, refused
	}

	// The controller makes the bucket (as it does at its start) and writes two objects: a small one and one of
	// several chunks, which needs the consumer's flow control.
	controller, _ := connect(ControllerName)
	cjs, err := jetstream.New(controller)
	if err != nil {
		t.Fatal(err)
	}
	if err := EnsureDeclarationFiles(ctx, cjs); err != nil {
		t.Fatalf("the controller cannot make %s: %v", DeclarationFilesBucket, err)
	}
	store, err := cjs.ObjectStore(ctx, DeclarationFilesBucket)
	if err != nil {
		t.Fatal(err)
	}
	put := func(content []byte) string {
		t.Helper()
		sum := sha256.Sum256(content)
		name := hex.EncodeToString(sum[:])
		if _, err := store.PutBytes(ctx, name, content); err != nil {
			t.Fatalf("the controller cannot write the object %s: %v", name[:8], err)
		}
		return name
	}
	small := bytes.Repeat([]byte("small file\n"), 100)
	large := make([]byte, 900<<10)
	for i := range large {
		large[i] = byte(i*7 + i>>8)
	}
	smallName, largeName := put(small), put(large)

	node, nodeRefused := connect("node.alpha")
	for name, want := range map[string][]byte{smallName: small, largeName: large} {
		got, err := nodeEnginesFileObject(ctx, node, name)
		if err != nil {
			t.Fatalf("a node-engine cannot read the object %s by its name: %v", name[:8], err)
		}
		if !bytes.Equal(got, want) {
			t.Fatalf("the node-engine read %d bytes of the object %s, want %d", len(got), name[:8], len(want))
		}
	}
	select {
	case err := <-nodeRefused:
		t.Fatalf("a node-engine's own read was refused something: %v", err)
	default:
	}

	// What each account is refused, judged by the server.
	listing := []byte(`{"subjects_filter":">"}`)
	bySeq := []byte(`{"seq":1}`)
	refusals := []struct {
		what string
		do   func(nc *nats.Conn)
	}{
		{"a list of the bucket (stream info asking for every subject)", func(nc *nats.Conn) {
			_, _ = nc.Request("$JS.API.STREAM.INFO."+DeclarationFilesStream, listing, 300*time.Millisecond)
		}},
		{"a message get by sequence", func(nc *nats.Conn) {
			_, _ = nc.Request("$JS.API.STREAM.MSG.GET."+DeclarationFilesStream, bySeq, 300*time.Millisecond)
		}},
		{"a direct get by sequence", func(nc *nats.Conn) {
			_, _ = nc.Request("$JS.API.DIRECT.GET."+DeclarationFilesStream, bySeq, 300*time.Millisecond)
		}},
		{"a consumer on every object's name", func(nc *nats.Conn) {
			js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
			sub, err := js.SubscribeSync("$O."+DeclarationFilesBucket+".M.>", nats.OrderedConsumer(), nats.BindStream(DeclarationFilesStream))
			if err == nil {
				sub.Unsubscribe()
			}
		}},
		{"a write of an object's metadata", func(nc *nats.Conn) {
			_ = nc.Publish("$O."+DeclarationFilesBucket+".M.abc", []byte(`{}`))
			_ = nc.Flush()
		}},
		{"a write of a chunk", func(nc *nats.Conn) {
			_ = nc.Publish("$O."+DeclarationFilesBucket+".C.abc", []byte("x"))
			_ = nc.Flush()
		}},
	}
	refusedWith := func(who string, nc *nats.Conn, ch chan error, what string, do func(*nats.Conn)) {
		t.Helper()
		for len(ch) > 0 {
			<-ch
		}
		do(nc)
		_ = nc.Flush()
		deadline := time.After(3 * time.Second)
		for {
			select {
			case err := <-ch:
				if errors.Is(err, nats.ErrPermissionViolation) {
					return
				}
			case <-deadline:
				t.Errorf("the server did not refuse %s %s", who, what)
				return
			}
		}
	}
	for _, r := range refusals {
		nodeNC, nodeCh := connect("node.alpha")
		refusedWith("a node-engine", nodeNC, nodeCh, r.what, r.do)
		otherNC, otherCh := connect("alpha.other-module")
		refusedWith("another module", otherNC, otherCh, r.what, r.do)
	}
	// And another module cannot read an object by name either.
	other, otherCh := connect("alpha.other-module")
	refusedWith("another module", other, otherCh, "a read of one object by its name", func(nc *nats.Conn) {
		_, _ = nc.Request("$JS.API.DIRECT.GET."+DeclarationFilesStream+".$O."+DeclarationFilesBucket+".M."+
			base64.URLEncoding.EncodeToString([]byte(smallName)), nil, 300*time.Millisecond)
	})

	// Nothing was written by the refused: the bucket holds the two objects.
	var names int
	lister, err := store.List(ctx)
	if err != nil {
		t.Fatal(err)
	}
	names = len(lister)
	if names != 2 {
		t.Errorf("the bucket holds %d objects, want the 2 the controller wrote", names)
	}
}