317 lines
11 KiB
Go
317 lines
11 KiB
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, node, name string) ([]byte, error) {
|
|
js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stream, err := DeclarationFilesStreamOf(node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bucket, err := DeclarationFilesBucketOf(node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
meta, err := js.GetLastMsg(stream, "$O."+bucket+".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."+bucket+".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 buckets' 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 into each node's bucket, a node-engine's account reads one of its own bucket 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; it is refused every one of those and the read by name on
|
|
// another node's bucket, **the literal-`*` forms included** (a published subject `...M.*` or `...C.*` matches the grant's `*`: on
|
|
// its own bucket that reaches its own files, on another's it is refused); another module's account is refused all of them.
|
|
func TestNatsANodeEngineReadsAnObjectOfItsOwnBucketByNameAndNothingElse(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", "beta"}, 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(), 90*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 buckets (as it does at its start) and writes into each: for alpha a small object and one of
|
|
// several chunks, which needs the consumer's flow control; for beta one that holds a marker alpha must never read.
|
|
controller, _ := connect(ControllerName)
|
|
cjs, err := jetstream.New(controller)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil {
|
|
t.Fatalf("the controller cannot make the buckets: %v", err)
|
|
}
|
|
if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil {
|
|
t.Fatalf("asserting them again: %v", err)
|
|
}
|
|
put := func(node string, content []byte) string {
|
|
t.Helper()
|
|
store, err := cjs.ObjectStore(ctx, mustBucket(t, node))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
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 of %s: %v", name[:8], node, 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)
|
|
}
|
|
secret := bytes.Repeat([]byte("BETA-SECRET "), 50)
|
|
smallName, largeName, betaName := put("alpha", small), put("alpha", large), put("beta", secret)
|
|
|
|
alpha, alphaRefused := connect("node.alpha")
|
|
for name, want := range map[string][]byte{smallName: small, largeName: large} {
|
|
got, err := nodeEnginesFileObject(ctx, alpha, "alpha", name)
|
|
if err != nil {
|
|
t.Fatalf("a node-engine cannot read the object %s of its own bucket 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))
|
|
}
|
|
}
|
|
beta, _ := connect("node.beta")
|
|
if got, err := nodeEnginesFileObject(ctx, beta, "beta", betaName); err != nil || !bytes.Equal(got, secret) {
|
|
t.Fatalf("beta's node-engine reads its own object: %v", err)
|
|
}
|
|
select {
|
|
case err := <-alphaRefused:
|
|
t.Fatalf("a node-engine's own read was refused something: %v", err)
|
|
default:
|
|
}
|
|
|
|
// What each account is refused, judged by the server.
|
|
alphaStream, alphaBucket := mustStream(t, "alpha"), mustBucket(t, "alpha")
|
|
betaStream, betaBucket := mustStream(t, "beta"), mustBucket(t, "beta")
|
|
listing := []byte(`{"subjects_filter":">"}`)
|
|
bySeq := []byte(`{"seq":1}`)
|
|
type refusal struct {
|
|
what string
|
|
do func(nc *nats.Conn)
|
|
}
|
|
onBucket := func(stream, bucket, which string) []refusal {
|
|
return []refusal{
|
|
{which + ": a list of the bucket (stream info asking for every subject)", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.STREAM.INFO."+stream, listing, 300*time.Millisecond)
|
|
}},
|
|
{which + ": a message get by sequence", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.STREAM.MSG.GET."+stream, bySeq, 300*time.Millisecond)
|
|
}},
|
|
{which + ": a direct get by sequence", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.DIRECT.GET."+stream, bySeq, 300*time.Millisecond)
|
|
}},
|
|
{which + ": a consumer on every object's name", func(nc *nats.Conn) {
|
|
js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
|
|
if sub, err := js.SubscribeSync("$O."+bucket+".M.>", nats.OrderedConsumer(), nats.BindStream(stream)); err == nil {
|
|
sub.Unsubscribe()
|
|
}
|
|
}},
|
|
{which + ": a write of an object's metadata", func(nc *nats.Conn) {
|
|
_ = nc.Publish("$O."+bucket+".M.abc", []byte(`{}`))
|
|
_ = nc.Flush()
|
|
}},
|
|
{which + ": a write of a chunk", func(nc *nats.Conn) {
|
|
_ = nc.Publish("$O."+bucket+".C.abc", []byte("x"))
|
|
_ = nc.Flush()
|
|
}},
|
|
}
|
|
}
|
|
refusedWith := func(who string, nc *nats.Conn, ch chan error, r refusal) {
|
|
t.Helper()
|
|
for len(ch) > 0 {
|
|
<-ch
|
|
}
|
|
r.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, r.what)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
// Alpha's account on its own bucket (list, sequence reads, names, writes) and on beta's (all of that, and the read by name,
|
|
// and the literal-`*` probes of the review).
|
|
reachBeta := []refusal{
|
|
{"beta's bucket: a read of one object by its name", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.DIRECT.GET."+betaStream+".$O."+betaBucket+".M."+
|
|
base64.URLEncoding.EncodeToString([]byte(betaName)), nil, 300*time.Millisecond)
|
|
}},
|
|
{"beta's bucket: a direct get with a literal `*` for the name", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.DIRECT.GET."+betaStream+".$O."+betaBucket+".M.*", nil, 300*time.Millisecond)
|
|
}},
|
|
{"beta's bucket: a consumer with a literal `*` for the chunks", func(nc *nats.Conn) {
|
|
js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
|
|
if sub, err := js.SubscribeSync("$O."+betaBucket+".C.*", nats.OrderedConsumer(), nats.BindStream(betaStream)); err == nil {
|
|
sub.Unsubscribe()
|
|
}
|
|
}},
|
|
}
|
|
all := append(append(onBucket(alphaStream, alphaBucket, "its own bucket"), onBucket(betaStream, betaBucket, "beta's bucket")...), reachBeta...)
|
|
for _, r := range all {
|
|
nodeNC, nodeCh := connect("node.alpha")
|
|
refusedWith("a node-engine", nodeNC, nodeCh, r)
|
|
}
|
|
// Another module is refused all of that on both buckets, and the reads by name on both.
|
|
for _, r := range append(append(onBucket(alphaStream, alphaBucket, "alpha's bucket"), reachBeta...),
|
|
refusal{"alpha's bucket: a read of one object by its name", func(nc *nats.Conn) {
|
|
_, _ = nc.Request("$JS.API.DIRECT.GET."+alphaStream+".$O."+alphaBucket+".M."+
|
|
base64.URLEncoding.EncodeToString([]byte(smallName)), nil, 300*time.Millisecond)
|
|
}}) {
|
|
otherNC, otherCh := connect("alpha.other-module")
|
|
refusedWith("another module", otherNC, otherCh, r)
|
|
}
|
|
|
|
// The literal `*` on alpha's own bucket reaches alpha's own files and never beta's: the chunks it is given hold no marker of beta's.
|
|
js, err := alpha.JetStream(nats.MaxWait(2 * time.Second))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if sub, err := js.SubscribeSync("$O."+alphaBucket+".C.*", nats.OrderedConsumer(), nats.BindStream(alphaStream)); err == nil {
|
|
for i := 0; i < 20; i++ {
|
|
msg, err := sub.NextMsg(300 * time.Millisecond)
|
|
if err != nil {
|
|
break
|
|
}
|
|
if bytes.Contains(msg.Data, []byte("BETA-SECRET")) {
|
|
t.Fatal("a literal `*` on alpha's bucket gave it a chunk of beta's file")
|
|
}
|
|
}
|
|
sub.Unsubscribe()
|
|
}
|
|
|
|
// Nothing was written by the refused: each bucket holds what the controller wrote.
|
|
for node, want := range map[string]int{"alpha": 2, "beta": 1} {
|
|
store, err := cjs.ObjectStore(ctx, mustBucket(t, node))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
infos, err := store.List(ctx)
|
|
if err != nil || len(infos) != want {
|
|
t.Errorf("the bucket of %s holds %d objects (%v), want the %d the controller wrote", node, len(infos), err, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func mustBucket(t *testing.T, node string) string {
|
|
t.Helper()
|
|
b, err := DeclarationFilesBucketOf(node)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func mustStream(t *testing.T, node string) string {
|
|
t.Helper()
|
|
b, err := DeclarationFilesStreamOf(node)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return b
|
|
}
|