394 lines
13 KiB
Go
394 lines
13 KiB
Go
package broker
|
|
|
|
import (
|
|
"context"
|
|
"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"
|
|
)
|
|
|
|
// **On a real server, as composed** (design 57 package P3, issue 669): a machine's provisioner client reads and
|
|
// watches its own machine's keys in a domain's bucket, and the server refuses it another machine's keys, a write,
|
|
// and a report or a vault ask in another's name.
|
|
func TestAServerComposedFromTheGrantsRefusesTheProvisionerClientAnotherMachinesKeys(t *testing.T) {
|
|
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
vault := Seat{Name: "mesh-vault", Scope: "mesh", Serves: []string{"get"}}
|
|
accounts, err := ComposeAccounts([]Principal{
|
|
{Kind: KindController, PasswordHash: string(hash)},
|
|
{Kind: KindModule, Node: "novox", Module: "mesh-config", State: []string{"resolved"}, PasswordHash: string(hash)},
|
|
{Kind: KindModule, Node: "novox", Module: "mesh-vault", Holds: []Seat{vault}, PasswordHash: string(hash)},
|
|
{Kind: KindModule, Node: "novox", Module: "postgres", State: []string{"bound"}, PasswordHash: string(hash)},
|
|
{Kind: KindProvisioner, Node: "g14", Module: "mesh-provisioner", BoundBuckets: []string{"postgres_bound"},
|
|
PasswordHash: string(hash)},
|
|
})
|
|
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")
|
|
}
|
|
defer s.Shutdown()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
connect := func(user string, refusals chan error) *nats.Conn {
|
|
t.Helper()
|
|
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
|
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
|
|
if refusals != nil {
|
|
refusals <- err
|
|
}
|
|
}))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
return nc
|
|
}
|
|
|
|
// The controller makes the buckets, as it does from registration; the controller writes its mirror and
|
|
// mesh-config writes its own bucket.
|
|
control, err := jetstream.New(connect("controller", nil))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
put := func(js jetstream.JetStream, bucket string, values map[string]string) {
|
|
t.Helper()
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for k, v := range values {
|
|
if _, err := kv.PutString(ctx, k, v); err != nil {
|
|
t.Fatalf("writing %s in %s: %v", k, bucket, err)
|
|
}
|
|
}
|
|
}
|
|
for _, bucket := range append(append([]string{"postgres_bound"}, ProvisionerReads...), ProvisionerReadsWhole...) {
|
|
if _, err := control.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
put(control, AssignmentsBucket, map[string]string{"g14.dunst": "{}", "shanks.dunst": "{}"})
|
|
provider, err := jetstream.New(connect("novox.postgres", nil))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
put(provider, "postgres_bound", map[string]string{"g14.gitea.postgres-database": "{}",
|
|
"shanks.photos.postgres-database": "{}"})
|
|
config, err := jetstream.New(connect("novox.mesh-config", nil))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
put(config, ConfigResolvedBucket, map[string]string{"g14.dunst.font-size": "13", "shanks.dunst.font-size": "11"})
|
|
|
|
refusals := make(chan error, 16)
|
|
clientConn := connect("provisioner.g14", refusals)
|
|
client, err := jetstream.New(clientConn)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
resolved, err := client.KeyValue(ctx, ConfigResolvedBucket)
|
|
if err != nil {
|
|
t.Fatalf("the client cannot bind to the config domain's bucket: %v", err)
|
|
}
|
|
|
|
// Its own machine's key, read and watched; the watch delivers the current value first.
|
|
if e, err := resolved.Get(ctx, "g14.dunst.font-size"); err != nil || string(e.Value()) != "13" {
|
|
t.Fatalf("its own machine's key: %v %v", e, err)
|
|
}
|
|
w, err := resolved.Watch(ctx, "g14.>")
|
|
if err != nil {
|
|
t.Fatalf("watching its own machine's keys: %v", err)
|
|
}
|
|
select {
|
|
case e := <-w.Updates():
|
|
if e == nil || e.Key() != "g14.dunst.font-size" {
|
|
t.Fatalf("the watch delivered %v first", e)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("the watch of its own machine's keys delivered nothing")
|
|
}
|
|
_ = w.Stop()
|
|
assignments, err := client.KeyValue(ctx, AssignmentsBucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := assignments.Get(ctx, "g14.dunst"); err != nil {
|
|
t.Fatalf("its own machine's placement: %v", err)
|
|
}
|
|
|
|
// Another machine's key: refused by the server, read or watched, and every key at once too; and a write of
|
|
// a domain's bucket, which is the domain's alone. Each refusal is the server's word, not a missing key.
|
|
drain := func() {
|
|
for {
|
|
select {
|
|
case <-refusals:
|
|
default:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
serverRefuses := func(what string, act func(context.Context) error) {
|
|
t.Helper()
|
|
drain()
|
|
short, done := context.WithTimeout(ctx, 2*time.Second)
|
|
defer done()
|
|
if err := act(short); err == nil {
|
|
t.Errorf("the client may %s", what)
|
|
}
|
|
select {
|
|
case err := <-refusals:
|
|
if !errors.Is(err, nats.ErrPermissionViolation) {
|
|
t.Errorf("%s: the server said %v, want a permissions violation", what, err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Errorf("%s: the server did not refuse it", what)
|
|
}
|
|
}
|
|
serverRefuses("read another machine's config", func(c context.Context) error {
|
|
_, err := resolved.Get(c, "shanks.dunst.font-size")
|
|
return err
|
|
})
|
|
// Another machine's placement is read, since a provider's consumers are bound to it in their own machines'
|
|
// plans (ADR 0373, issue 679), and never written.
|
|
if _, err := assignments.Get(ctx, "shanks.dunst"); err != nil {
|
|
t.Fatalf("another machine's plan, which binds its consumers to a provider here: %v", err)
|
|
}
|
|
serverRefuses("write the assignments mirror", func(c context.Context) error {
|
|
_, err := assignments.PutString(c, "shanks.dunst", "{}")
|
|
return err
|
|
})
|
|
// A provider's entries: its own machine's consumers' keys alone (issue 679).
|
|
entries, err := client.KeyValue(ctx, "postgres_bound")
|
|
if err != nil {
|
|
t.Fatalf("the client cannot bind to a provider's bucket of entries: %v", err)
|
|
}
|
|
if _, err := entries.Get(ctx, "g14.gitea.postgres-database"); err != nil {
|
|
t.Fatalf("its own machine's consumer's entry: %v", err)
|
|
}
|
|
serverRefuses("read another machine's consumer's entry", func(c context.Context) error {
|
|
_, err := entries.Get(c, "shanks.photos.postgres-database")
|
|
return err
|
|
})
|
|
serverRefuses("write a provider's entry", func(c context.Context) error {
|
|
_, err := entries.PutString(c, "g14.gitea.postgres-database", "{}")
|
|
return err
|
|
})
|
|
for _, keys := range []string{"shanks.>", ">"} {
|
|
serverRefuses("watch "+keys, func(c context.Context) error {
|
|
w, err := resolved.Watch(c, keys)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = w.Stop() }()
|
|
select {
|
|
case e := <-w.Updates():
|
|
if e != nil {
|
|
return nil
|
|
}
|
|
return errors.New("nothing delivered")
|
|
case <-c.Done():
|
|
return c.Err()
|
|
}
|
|
})
|
|
}
|
|
serverRefuses("write the config domain's bucket", func(c context.Context) error {
|
|
_, err := resolved.PutString(c, "g14.dunst.font-size", "40")
|
|
return err
|
|
})
|
|
|
|
// What it applied, on its own machine and no other's; the vault asked in its own name and no other's.
|
|
refused := func(subject string, want bool) {
|
|
t.Helper()
|
|
drain()
|
|
if err := clientConn.Publish(subject, []byte(`{}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = clientConn.Flush()
|
|
select {
|
|
case err := <-refusals:
|
|
if !want {
|
|
t.Errorf("%s was refused: %v", subject, err)
|
|
} else if !errors.Is(err, nats.ErrPermissionViolation) {
|
|
t.Errorf("%s: the server said %v, want a permissions violation", subject, err)
|
|
}
|
|
case <-time.After(500 * time.Millisecond):
|
|
if want {
|
|
t.Errorf("the server did not refuse %s", subject)
|
|
}
|
|
}
|
|
}
|
|
refused("prov.applied.g14.dunst", false)
|
|
refused("prov.applied.shanks.dunst", true)
|
|
refused("mesh.seat.mesh-vault.call.get.provisioner~g14", false)
|
|
refused("mesh.seat.mesh-vault.call.get.g14~node-tools", true)
|
|
refused("mesh.seat.mesh-vault.call.accept.provisioner~g14", true)
|
|
refused("mesh.mod.dunst.tool.anything", true)
|
|
}
|
|
|
|
// **On a real server, as composed** (novox/hq ADR 0370 §3, issue 685): a machine's tool runner asks its own
|
|
// machine's provisioner client and is answered; the server refuses the ask from another machine's tool runner, from
|
|
// a module's own account and from the client of another machine, and refuses the client what another machine's
|
|
// tool runner would hear.
|
|
func TestAServerComposedFromTheGrantsLetsOnlyItsOwnToolRunnerAskTheProvisionerClient(t *testing.T) {
|
|
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
accounts, err := ComposeAccounts([]Principal{
|
|
{Kind: KindController, PasswordHash: string(hash)},
|
|
{Kind: KindNodeTools, Node: "g14", Module: RuntimeModule, PasswordHash: string(hash),
|
|
Carries: []Declared{{Module: "claude-code"}}},
|
|
{Kind: KindNodeTools, Node: "shanks", Module: RuntimeModule, PasswordHash: string(hash),
|
|
Carries: []Declared{{Module: "claude-code"}}},
|
|
{Kind: KindModule, Node: "g14", Module: "dunst", PasswordHash: string(hash)},
|
|
{Kind: KindProvisioner, Node: "g14", Module: "mesh-provisioner", BoundBuckets: []string{"postgres_bound"},
|
|
PasswordHash: string(hash)},
|
|
{Kind: KindProvisioner, Node: "shanks", Module: "mesh-provisioner", PasswordHash: string(hash)},
|
|
})
|
|
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\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")
|
|
}
|
|
defer s.Shutdown()
|
|
connect := func(user string, refusals chan error) *nats.Conn {
|
|
t.Helper()
|
|
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
|
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
|
|
if refusals != nil {
|
|
refusals <- err
|
|
}
|
|
}))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
return nc
|
|
}
|
|
|
|
clientRefusals := make(chan error, 8)
|
|
client := connect("provisioner.g14", clientRefusals)
|
|
asked := make(chan string, 8)
|
|
if _, err := client.Subscribe(ProvisionerAskSubject("g14"), func(m *nats.Msg) {
|
|
asked <- string(m.Data)
|
|
_ = m.Respond([]byte(`{"value":"answered"}`))
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := client.Flush(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case err := <-clientRefusals:
|
|
t.Fatalf("the client may not hear its own machine's asks: %v", err)
|
|
case <-time.After(300 * time.Millisecond):
|
|
}
|
|
|
|
// Its own machine's tool runner asks and is answered.
|
|
runner := connect("g14.node-tools", nil)
|
|
reply, err := runner.Request(ProvisionerAskSubject("g14"), []byte(`{"module":"claude-code"}`), 5*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("g14's tool runner cannot ask g14's client: %v", err)
|
|
}
|
|
if string(reply.Data) != `{"value":"answered"}` {
|
|
t.Fatalf("answered %s", reply.Data)
|
|
}
|
|
<-asked
|
|
|
|
// Anybody else's ask is refused by the server, and never reaches the client.
|
|
for _, user := range []string{"shanks.node-tools", "g14.dunst", "provisioner.shanks", "controller"} {
|
|
refusals := make(chan error, 4)
|
|
nc := connect(user, refusals)
|
|
if err := nc.Publish(ProvisionerAskSubject("g14"), []byte(`{"module":"claude-code"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = nc.Flush()
|
|
select {
|
|
case err := <-refusals:
|
|
if !errors.Is(err, nats.ErrPermissionViolation) {
|
|
t.Errorf("%s: the server said %v, want a permissions violation", user, err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Errorf("the server did not refuse %s an ask of g14's client", user)
|
|
}
|
|
select {
|
|
case body := <-asked:
|
|
t.Errorf("g14's client heard %s from %s", body, user)
|
|
case <-time.After(200 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
// What moved is said under its own machine's name, heard by that machine's tool runner; another machine's name
|
|
// is refused it.
|
|
told := make(chan string, 4)
|
|
if _, err := runner.Subscribe("prov.changed.g14.>", func(m *nats.Msg) { told <- m.Subject }); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = runner.Flush()
|
|
if err := client.Publish(ProvisionerChangedSubject("g14", "claude-code"), []byte(`{"kind":"config"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case subject := <-told:
|
|
if subject != "prov.changed.g14.claude-code" {
|
|
t.Errorf("heard %s", subject)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Error("g14's tool runner did not hear what g14's client said moved")
|
|
}
|
|
if err := client.Publish(ProvisionerChangedSubject("shanks", "claude-code"), []byte(`{}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = client.Flush()
|
|
select {
|
|
case err := <-clientRefusals:
|
|
if !errors.Is(err, nats.ErrPermissionViolation) {
|
|
t.Errorf("the server said %v, want a permissions violation", err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Error("the server did not refuse g14's client a word in shanks's name")
|
|
}
|
|
}
|