//go:build integration package buscompose import ( "bytes" "context" "encoding/base64" "encoding/json" "os" "path/filepath" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/seal" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) // Defends novox/hq ADR 0547 clause 7 on a real server (issue 1945): with the list this composer writes, the grants part // creates its object bucket and its state, puts the sealed list and names it in `user-list.sealed`; the node-engine of // the node holding mesh-broker watches that one key, reads the object by its name and opens it with its key, and may // neither list the bucket, nor read another key of the state, nor write; another node reads nothing of either. func TestTheBusNodeFollowsTheGrantsPartsSealedListUnderTheComposedGrants(t *testing.T) { records := Records{ Nodes: []string{"novox", "ace"}, Assigned: map[string][]Declared{ "novox": {{Module: "nats", SnapshotsTheBus: true}, {Module: GrantsModule, State: []Bucket{{Module: GrantsModule, Name: "user-list"}}}}, "ace": {{Module: "plex"}}, }, } users, err := Users(records) if err != nil { t.Fatal(err) } for i := range users { users[i].PasswordHash = "pw-" + users[i].Username() } accounts, err := ComposeAccounts(users) if err != nil { t.Fatal(err) } dir := t.TempDir() conf := filepath.Join(dir, "server.conf") if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: \""+dir+"\" }\n"+accounts), 0o600); err != nil { t.Fatal(err) } opts, err := server.ProcessConfigFile(conf) if err != nil { t.Fatalf("the composed list is not the server's: %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(30 * time.Second) { t.Fatal("the bus did not come up") } t.Cleanup(s.Shutdown) dial := func(user string) *nats.Conn { nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw-"+user), nats.CustomInboxPrefix("_INBOX."+user)) if err != nil { t.Fatalf("%s: %v", user, err) } t.Cleanup(nc.Close) return nc } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() // The grants part: its state and its object bucket, the sealed list put and named. grants, err := jetstream.New(dial("novox." + GrantsModule)) if err != nil { t.Fatal(err) } // Its state is made as the grants part makes a declared state, by its stream's name (CreateKeyValue would ask the // account's info, which no module is granted). if _, err := grants.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_" + GrantsUserListBucket, Subjects: []string{"$KV." + GrantsUserListBucket + ".>"}, MaxMsgsPerSubject: 1, AllowDirect: true, AllowRollup: true, DenyDelete: true, Discard: jetstream.DiscardNew}); err != nil { t.Fatalf("the grants part's state: %v", err) } kv, err := grants.KeyValue(ctx, GrantsUserListBucket) if err != nil { t.Fatalf("the grants part binds its state: %v", err) } objects, err := grants.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: GrantsUserListObjects}) if err != nil { t.Fatalf("the grants part's object bucket: %v", err) } node, err := seal.GenerateKeyPair() if err != nil { t.Fatal(err) } list := []byte(accounts) digest := delivery.UserListDigest(list) object, err := delivery.SealUserList(list, node.PublicKey) if err != nil { t.Fatal(err) } name := delivery.UserListObjectName(digest) for range 2 { // put twice: a second put of a name purges its old chunks if _, err := objects.PutBytes(ctx, name, object); err != nil { t.Fatalf("the grants part put its object: %v", err) } } value, err := delivery.EncodeGrantsUserList(delivery.GrantsUserList{Follow: delivery.FollowGrants, Digest: digest, Object: name, SealedTo: delivery.SealingKeyID(node.PublicKey), Node: "novox", Revision: 1, Users: len(users)}) if err != nil { t.Fatal(err) } if _, err := kv.Put(ctx, GrantsUserListKey, value); err != nil { t.Fatal(err) } if _, err := kv.Put(ctx, "list", []byte(`{"v":1}`)); err != nil { t.Fatal(err) } // The bus node's engine: the key watched, the object by name (as mesh-host's link reads a declaration's file). busNode := dial("node.novox") js, err := busNode.JetStream(nats.MaxWait(5 * time.Second)) if err != nil { t.Fatal(err) } keys, err := js.KeyValue(GrantsUserListBucket) if err != nil { t.Fatalf("the bus node binds the grants part's state: %v", err) } w, err := keys.Watch(GrantsUserListKey) if err != nil { t.Fatalf("the bus node watches the key: %v", err) } defer w.Stop() var got delivery.GrantsUserList select { case e := <-w.Updates(): if e == nil { t.Fatal("the watch found no value") } if got, err = delivery.DecodeGrantsUserList(e.Value()); err != nil { t.Fatal(err) } case <-ctx.Done(): t.Fatal("the watch said nothing") } read := func(js nats.JetStreamContext, name string) ([]byte, error) { meta, err := js.GetLastMsg(GrantsUserListObjectStream, "$O."+GrantsUserListObjects+".M."+ base64.URLEncoding.EncodeToString([]byte(name)), nats.DirectGet()) if err != nil { return nil, err } var info struct { NUID string `json:"nuid"` Size int `json:"size"` } if err := json.Unmarshal(meta.Data, &info); err != nil { return nil, err } sub, err := js.SubscribeSync("$O."+GrantsUserListObjects+".C."+info.NUID, nats.OrderedConsumer(), nats.BindStream(GrantsUserListObjectStream)) if err != nil { return nil, err } defer sub.Unsubscribe() var body []byte for len(body) < info.Size { m, err := sub.NextMsg(5 * time.Second) if err != nil { return nil, err } body = append(body, m.Data...) } return body, nil } body, err := read(js, got.Object) if err != nil { t.Fatalf("the bus node reads the object by its name: %v", err) } opened, err := delivery.OpenUserList(body, got.Digest, func(b seal.Box) ([]byte, error) { s, err := seal.Open(b, node.PrivateKey) return []byte(s), err }) if err != nil || !bytes.Equal(opened, list) { t.Fatalf("the bus node opened %d bytes, %v", len(opened), err) } // And nothing beyond: no other key, no list of the bucket, no write. if _, err := keys.Get("list"); err == nil { t.Error("the bus node read another key of the grants part's state") } if _, err := js.StreamInfo(GrantsUserListObjectStream); err == nil { t.Error("the bus node read the object stream's info, which lists every name") } if _, err := js.Publish("$O."+GrantsUserListObjects+".M.x", []byte("x")); err == nil { t.Error("the bus node wrote the grants part's bucket") } // Another node reads nothing of it. other, err := dial("node.ace").JetStream(nats.MaxWait(2 * time.Second)) if err != nil { t.Fatal(err) } if _, err := read(other, got.Object); err == nil { t.Error("another node read the object") } if _, err := other.GetLastMsg("KV_"+GrantsUserListBucket, "$KV."+GrantsUserListBucket+"."+GrantsUserListKey, nats.DirectGet()); err == nil { t.Errorf("another node read the key: %v", err) } }