diff --git a/scenarios/signed-logins.yml b/scenarios/signed-logins.yml new file mode 100644 index 0000000..e916746 --- /dev/null +++ b/scenarios/signed-logins.yml @@ -0,0 +1,29 @@ +# Research 053 question 8: three machines for the signed-login bus. +# a = today's style server (static users, JetStream), b and c = a cluster in operator mode with a +# full resolver, a joined to b as a leaf. No overlay and no mesh software: the bed brings its own +# nats-server (the program test/bed-programs/signed-logins embeds 2.11.17). +scenario: signed-logins + +segments: + hosting: + kind: public + cidr: [192.0.2.0/24] + +machines: + a: + at: { segment: hosting, address: [192.0.2.11] } + inbound: allow + memory: 2GiB + cpus: 2 + b: + at: { segment: hosting, address: [192.0.2.12] } + inbound: allow + memory: 2GiB + cpus: 2 + c: + at: { segment: hosting, address: [192.0.2.13] } + inbound: allow + memory: 2GiB + cpus: 2 + +snapshot: raised diff --git a/test/bed-programs/signed-logins/drive.go b/test/bed-programs/signed-logins/drive.go new file mode 100644 index 0000000..1ecebad --- /dev/null +++ b/test/bed-programs/signed-logins/drive.go @@ -0,0 +1,584 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "os" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/nats-io/nkeys" +) + +type env struct { + k Keys + op, sys, acc nkeys.KeyPair + sig nkeys.KeyPair + revoked map[string]time.Time + lastIat int64 + accountDoneAt time.Time +} + +func (e *env) url(ip string) string { return "nats://" + ip + ":4222" } + +func say(format string, a ...any) { + fmt.Printf("%s %s\n", time.Now().Format("15:04:05.000"), fmt.Sprintf(format, a...)) +} + +func ms(d time.Duration) string { return fmt.Sprintf("%.1f ms", float64(d.Microseconds())/1000) } + +type user struct { + kp nkeys.KeyPair + opt nats.Option +} + +func (e *env) newUser(name string) user { + kp := must(nkeys.CreateUser()) + j := userJWT(e.acc, e.sig, kp, name) + return user{kp, nats.UserJWT(func() (string, error) { return j, nil }, func(n []byte) ([]byte, error) { return kp.Sign(n) })} +} + +func (e *env) sysConn(ip string) *nats.Conn { + kp := must(nkeys.CreateUser()) + uc := newSysUser(e.sys, kp) + nc, err := nats.Connect(e.url(ip), nats.UserJWT(func() (string, error) { return uc, nil }, func(n []byte) ([]byte, error) { return kp.Sign(n) }), nats.Timeout(5*time.Second)) + if err != nil { + panic(fmt.Sprintf("system connection to %s: %v", ip, err)) + } + return nc +} + +// probe is one login attempt from this machine across the network. +func (e *env) probe(ip string, u user) error { + nc, err := nats.Connect(e.url(ip), u.opt, nats.NoReconnect(), nats.Timeout(3*time.Second)) + if err == nil { + nc.Close() + } + return err +} + +type open struct { + nc *nats.Conn + closed atomic.Int64 +} + +func (e *env) openConn(ip string, u user) *open { + o := &open{} + nc, err := nats.Connect(e.url(ip), u.opt, nats.NoReconnect(), nats.Timeout(3*time.Second), + nats.ClosedHandler(func(*nats.Conn) { o.closed.Store(time.Now().UnixNano()) })) + if err != nil { + panic(fmt.Sprintf("open connection to %s: %v", ip, err)) + } + nc.Flush() + o.nc = nc + return o +} + +// push sends an account JWT to one server's resolver over the system account. The reply is not proof it applied. +func (e *env) push(sys *nats.Conn, j string) (string, time.Duration) { + t0 := time.Now() + m, err := sys.Request("$SYS.REQ.ACCOUNT."+pubOf(e.acc)+".CLAIMS.UPDATE", []byte(j), 5*time.Second) + if err != nil { + return "error: " + err.Error(), time.Since(t0) + } + return string(m.Data), time.Since(t0) +} + +// nextAccountJWT builds the account JWT with the revocation list; its iat has to be a later second than the last one. +func (e *env) nextAccountJWT() string { + for time.Now().Unix() <= e.lastIat { + time.Sleep(50 * time.Millisecond) + } + e.lastIat = time.Now().Unix() + return accountJWT(e.op, e.acc, e.sig, e.revoked) +} + +func (e *env) refusedAfter(ip string, u user, t0 time.Time, limit time.Duration) (time.Duration, int) { + n := 0 + for time.Since(t0) < limit { + if err := e.probe(ip, u); err != nil { + return time.Since(t0), n + } + n++ + time.Sleep(time.Millisecond) + } + return -1, n +} + +func newSysUser(sys, ukp nkeys.KeyPair) string { return userClaimsFor(sys, ukp) } + +func drive(args []string) { + fs := flag.NewFlagSet("drive", flag.ExitOnError) + keysFile := fs.String("keys", "/root/q8/keys.json", "keys") + only := fs.String("only", "2,4,3", "which parts, in order") + trials := fs.Int("trials", 3, "trials per part") + fs.Parse(args) + raw, err := os.ReadFile(*keysFile) + if err != nil { + panic(err) + } + e := &env{revoked: map[string]time.Time{}} + json.Unmarshal(raw, &e.k) + e.op, e.sys, e.acc, e.sig = fromSeed(e.k.Op), fromSeed(e.k.Sys), fromSeed(e.k.Acc), fromSeed(e.k.Sig) + e.lastIat = time.Now().Unix() + // the account JWT the servers were preloaded with has an iat from gen time; ours must be later + time.Sleep(1100 * time.Millisecond) + if d := waitRoutes(e, 1, 90*time.Second); d < 0 { + say("the route between B and C did not come up") + } else { + say("cluster formed: route B-C up after %s", ms(d)) + } + for _, p := range strings.Split(*only, ",") { + switch p { + case "2": + part2(e, *trials) + case "3": + part3(e, *trials) + case "4": + part4(e) + } + } + say("DRIVE-DONE") +} + +func routes(ip string) int { + r, err := http.Get("http://" + ip + ":8222/routez") + if err != nil { + return -1 + } + defer r.Body.Close() + var v struct { + NumRoutes int `json:"num_routes"` + } + json.NewDecoder(r.Body).Decode(&v) + return v.NumRoutes +} + +func waitRoutes(e *env, want int, limit time.Duration) time.Duration { + t0 := time.Now() + for time.Since(t0) < limit { + if routes(e.k.B) == want && routes(e.k.C) == want { + return time.Since(t0) + } + time.Sleep(100 * time.Millisecond) + } + return -1 +} + +func agent(e *env, what string) (string, time.Time) { + r, err := http.Get("http://" + e.k.C + ":9099/" + what) + t := time.Now() + if err != nil { + return "error: " + err.Error(), t + } + defer r.Body.Close() + b, _ := io.ReadAll(r.Body) + return string(b), t +} + +// ---- part 2: revocation across the network ---- + +func part2(e *env, n int) { + say("=== PART 2: revocation over the network (B and C, pushed to B from A)") + sysB := e.sysConn(e.k.B) + defer sysB.Close() + for run := 1; run <= n; run++ { + u := e.newUser(fmt.Sprintf("p2-%d", run)) + ctl := e.newUser(fmt.Sprintf("p2-ctl-%d", run)) + ob, oc := e.openConn(e.k.B, u), e.openConn(e.k.C, u) + cb, cc := e.openConn(e.k.B, ctl), e.openConn(e.k.C, ctl) + e.revoked[pubOf(u.kp)] = time.Now() + j := e.nextAccountJWT() + t0 := time.Now() + reply, rtt := e.push(sysB, j) + type res struct { + ip string + refused time.Duration + acc int + } + var wg sync.WaitGroup + out := make([]res, 2) + for i, ip := range []string{e.k.B, e.k.C} { + wg.Add(1) + go func() { + defer wg.Done() + d, a := e.refusedAfter(ip, u, t0, 30*time.Second) + out[i] = res{ip, d, a} + }() + } + wg.Wait() + time.Sleep(500 * time.Millisecond) + cl := func(o *open) string { + if v := o.closed.Load(); v != 0 { + return ms(time.Duration(v - t0.UnixNano())) + } + return "NOT closed" + } + alive := func(o *open) bool { return o.nc.IsConnected() } + say("RESULT part2 run %d: push reply %q rtt %s; new login refused after: B %s (%d accepted before), C %s (%d accepted before); open connection closed after: B %s, C %s; control user still connected: B %v, C %v", + run, strings.TrimSpace(reply)[:min(len(strings.TrimSpace(reply)), 60)], ms(rtt), ms(out[0].refused), out[0].acc, ms(out[1].refused), out[1].acc, cl(ob), cl(oc), alive(cb), alive(cc)) + for _, o := range []*open{ob, oc, cb, cc} { + o.nc.Close() + } + } +} + +// ---- part 3: C cut off while a user is revoked ---- + +type seen struct { + at time.Time + subject string + data []byte +} + +func part3(e *env, n int) { + say("=== PART 3: C cut off from B while a user is revoked; modes a (periodic sync only) and b (re-push on C's return)") + for _, mode := range []string{"a", "b"} { + for run := 1; run <= n; run++ { + trial3(e, mode, run) + } + } +} + +func trial3(e *env, mode string, run int) { + say("--- trial %s%d", mode, run) + if d := waitRoutes(e, 1, 90*time.Second); d < 0 { + say("RESULT part3 %s%d: the route B-C did not come up (routes B=%d C=%d); trial skipped", mode, run, routes(e.k.B), routes(e.k.C)) + return + } + time.Sleep(3 * time.Second) + sysB, sysC := e.sysConn(e.k.B), e.sysConn(e.k.C) + defer sysB.Close() + defer sysC.Close() + u, ctl := e.newUser(fmt.Sprintf("p3-%s%d", mode, run)), e.newUser(fmt.Sprintf("p3-ctl-%s%d", mode, run)) + oc, cc := e.openConn(e.k.C, u), e.openConn(e.k.C, ctl) + defer oc.nc.Close() + defer cc.nc.Close() + var mu sync.Mutex + var events []seen + sysB.Subscribe("$SYS.>", func(m *nats.Msg) { + mu.Lock() + events = append(events, seen{time.Now(), m.Subject, m.Data}) + mu.Unlock() + }) + sysB.Flush() + + r, tCut := agent(e, "block") + say("cut: agent said %q", r) + dr := waitRoutes(e, 0, 60*time.Second) + say("routes gone on B and C after %s (ping interval 2s, 2 missed pings)", ms(dr)) + time.Sleep(1100 * time.Millisecond) + e.revoked[pubOf(u.kp)] = time.Now() + j := e.nextAccountJWT() + tPush := time.Now() + reply, rtt := e.push(sysB, j) + dB, _ := e.refusedAfter(e.k.B, u, tPush, 10*time.Second) + time.Sleep(4 * time.Second) + accC := e.probe(e.k.C, u) + say("partitioned: push to B reply %q rtt %s; B refused the user after %s; C still accepts the user: %v (cut lasted %s so far)", strings.TrimSpace(reply)[:min(len(strings.TrimSpace(reply)), 40)], ms(rtt), ms(dB), accC == nil, time.Since(tCut).Round(time.Second)) + + mu.Lock() + events = nil + mu.Unlock() + r, tHeal := agent(e, "unblock") + say("healed: agent said %q", r) + + var trigger atomic.Int64 // offset ns of the event used as trigger + var triggerSubj atomic.Value + var pushedRTT atomic.Int64 + var pushedReply atomic.Value + if mode == "b" { + go func() { + deadline := time.Now().Add(150 * time.Second) + for time.Now().Before(deadline) && trigger.Load() == 0 { + mu.Lock() + var hit *seen + for i := range events { + ev := events[i] + if strings.HasPrefix(ev.subject, "$SYS.SERVER.") && strings.HasSuffix(ev.subject, ".STATSZ") && ev.at.After(tHeal) { + var v struct { + Server struct{ Name string } `json:"server"` + } + if json.Unmarshal(ev.data, &v) == nil && v.Server.Name == "C" { + hit = &ev + break + } + } + } + mu.Unlock() + if hit != nil { + trigger.Store(int64(hit.at.Sub(tHeal))) + triggerSubj.Store(hit.subject) + rep, rt := e.push(sysC, j) + pushedRTT.Store(int64(rt)) + pushedReply.Store(rep) + return + } + time.Sleep(time.Millisecond) + } + }() + } + refused, accepted := e.refusedAfter(e.k.C, u, tHeal, 150*time.Second) + time.Sleep(700 * time.Millisecond) + closedAt := "NOT closed" + if v := oc.closed.Load(); v != 0 { + closedAt = ms(time.Duration(v - tHeal.UnixNano())) + } + line := fmt.Sprintf("RESULT part3 %s%d: after the link came back C accepted the revoked user for %s (%d logins accepted), open connection closed after %s, control user still connected %v", + mode, run, ms(refused), accepted, closedAt, cc.nc.IsConnected()) + if mode == "b" { + if t := trigger.Load(); t != 0 { + subj, _ := triggerSubj.Load().(string) + rep, _ := pushedReply.Load().(string) + line += fmt.Sprintf("; trigger: first STATSZ from server C arrived on B's system account %s after the link came back (%s), push to C reply %q rtt %s", + ms(time.Duration(t)), strings.ReplaceAll(subj, "$SYS.SERVER.", "$SYS.SERVER.."), strings.TrimSpace(rep)[:min(len(strings.TrimSpace(rep)), 30)], ms(time.Duration(pushedRTT.Load()))) + } else { + line += "; trigger NEVER fired" + } + } + say("%s", line) + if run == 1 { + mu.Lock() + first := map[string]time.Duration{} + count := map[string]int{} + for _, ev := range events { + k := normalise(ev.subject) + if _, ok := first[k]; !ok { + first[k] = ev.at.Sub(tHeal) + } + count[k]++ + } + mu.Unlock() + var ks []string + for k := range first { + ks = append(ks, k) + } + sort.Slice(ks, func(i, j int) bool { return first[ks[i]] < first[ks[j]] }) + say("system-account subjects seen on B after the link came back (trial %s%d), first-seen offset, count:", mode, run) + for _, k := range ks { + say(" %9s x%-3d %s", ms(first[k]), count[k], k) + } + } +} + +func normalise(s string) string { + parts := strings.Split(s, ".") + for i, p := range parts { + if len(p) >= 40 && strings.ToUpper(p) == p { + parts[i] = "" + } + } + return strings.Join(parts, ".") +} + +// ---- part 4: moving a stream and a KV bucket from A to B over the leaf link ---- + +func part4(e *env) { + say("=== PART 4: stream S (10000 msgs) and KV bucket cfg moved from A (legacy) to B/C (hub) over the leaf link, writer running") + ctx := context.Background() + lc, err := nats.Connect(e.url(e.k.A), nats.UserInfo("w", "pw")) + if err != nil { + panic(err) + } + defer lc.Close() + ljs := must(jetstream.New(lc)) + must(ljs.CreateStream(ctx, jetstream.StreamConfig{Name: "S", Subjects: []string{"mesh.s.>"}})) + t0 := time.Now() + for i := 0; i < 10000; i++ { + if _, err := ljs.Publish(ctx, fmt.Sprintf("mesh.s.%d", i%50), []byte(fmt.Sprintf("seed-%d", i)), jetstream.WithMsgID(fmt.Sprintf("seed-%d", i))); err != nil { + panic(err) + } + } + say("seeded S with 10000 messages on A in %s", time.Since(t0).Round(time.Millisecond)) + kv := must(ljs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "cfg", History: 5})) + for i := 0; i < 1000; i++ { + must(kv.Put(ctx, fmt.Sprintf("k%d", i), []byte(fmt.Sprintf("v%d", i)))) + } + for i := 0; i < 200; i++ { + for r := 0; r < 3; r++ { + must(kv.Put(ctx, fmt.Sprintf("k%d", i), []byte(fmt.Sprintf("v%d-r%d", i, r)))) + } + } + kv.Delete(ctx, "k999") + say("seeded KV cfg on A") + + for t := time.Now(); time.Since(t) < 60*time.Second; time.Sleep(500 * time.Millisecond) { + r, err := http.Get("http://" + e.k.B + ":8222/leafz") + if err != nil { + continue + } + var v struct { + Leafs int `json:"leafnodes"` + } + json.NewDecoder(r.Body).Decode(&v) + r.Body.Close() + if v.Leafs >= 2 { + say("leaf links from A to B up (%d connections: MESH and SYS)", v.Leafs) + break + } + } + u := e.newUser("p4") + hc, err := nats.Connect(e.url(e.k.B), u.opt) + if err != nil { + panic(err) + } + defer hc.Close() + hjs := must(jetstream.New(hc)) + ready := time.Now() + for time.Since(ready) < 60*time.Second { + if _, err := hjs.AccountInfo(ctx); err == nil { + break + } + time.Sleep(500 * time.Millisecond) + } + say("hub JetStream answered after %s", time.Since(ready).Round(time.Millisecond)) + + var stop atomic.Bool + var acked, kvacked, failed atomic.Int64 + done := make(chan bool) + go func() { + for i := 0; !stop.Load(); i++ { + if _, err := ljs.Publish(ctx, fmt.Sprintf("mesh.s.%d", i%50), []byte(fmt.Sprintf("live-%d", i)), jetstream.WithMsgID(fmt.Sprintf("live-%d", i))); err == nil { + acked.Add(1) + } else { + failed.Add(1) + } + if i%10 == 0 { + if _, err := kv.Put(ctx, fmt.Sprintf("live%d", i%20), []byte(fmt.Sprintf("x%d", i))); err == nil { + kvacked.Add(1) + } + } + time.Sleep(2 * time.Millisecond) + } + close(done) + }() + time.Sleep(300 * time.Millisecond) + + tm := time.Now() + for _, n := range []string{"S", "KV_cfg"} { + l := must(ljs.Stream(ctx, n)) + cfg := l.CachedInfo().Config + cfg.Subjects = nil + cfg.Sources = []*jetstream.StreamSource{{Name: n, External: &jetstream.ExternalStream{APIPrefix: "$JS.legacy.API"}}} + if _, err := hjs.CreateStream(ctx, cfg); err != nil { + panic(fmt.Sprintf("create source stream %s: %v", n, err)) + } + } + msgs := func(js jetstream.JetStream, n string) uint64 { + s, err := js.Stream(ctx, n) + if err != nil { + return 0 + } + i, err := s.Info(ctx) + if err != nil { + return 0 + } + return i.State.Msgs + } + waitEq := func(n string, limit time.Duration) time.Duration { + t := time.Now() + for time.Since(t) < limit { + if msgs(hjs, n) == msgs(ljs, n) && msgs(hjs, n) > 0 { + return time.Since(tm) + } + time.Sleep(20 * time.Millisecond) + } + return -1 + } + say("catch-up under writes, counts equal after: S %s, KV_cfg %s (the writer keeps going)", ms(waitEq("S", 60*time.Second)), ms(waitEq("KV_cfg", 60*time.Second))) + time.Sleep(3 * time.Second) + stop.Store(true) + <-done + say("writer stopped: S acked %d (failed %d), KV puts acked %d", acked.Load(), failed.Load(), kvacked.Load()) + tq := time.Now() + for _, n := range []string{"S", "KV_cfg"} { + for time.Since(tq) < 30*time.Second && msgs(hjs, n) != msgs(ljs, n) { + time.Sleep(10 * time.Millisecond) + } + } + say("quiesced: equal after %s; counts S legacy %d hub %d (expect %d); KV_cfg legacy %d hub %d", time.Since(tq).Round(time.Millisecond), + msgs(ljs, "S"), msgs(hjs, "S"), 10000+acked.Load(), msgs(ljs, "KV_cfg"), msgs(hjs, "KV_cfg")) + + // content: every message once on both sides + readAll := func(js jetstream.JetStream) (map[string]int, int) { + s := must(js.Stream(ctx, "S")) + info := must(s.Info(ctx)) + set := map[string]int{} + total := 0 + for seq := info.State.FirstSeq; seq <= info.State.LastSeq; seq++ { + m, err := s.GetMsg(ctx, seq) + if err != nil { + continue + } + set[string(m.Data)]++ + total++ + } + return set, total + } + ls, lt := readAll(ljs) + hs, ht := readAll(hjs) + dup, missing, extra := 0, 0, 0 + for k, c := range hs { + if c > 1 { + dup += c - 1 + } + if _, ok := ls[k]; !ok { + extra++ + } + } + for k := range ls { + if _, ok := hs[k]; !ok { + missing++ + } + } + say("RESULT part4 stream S: messages legacy %d (distinct %d) hub %d (distinct %d); on hub: duplicates %d, missing vs legacy %d, not on legacy %d; expected %d = 10000 seed + %d acked live", + lt, len(ls), ht, len(hs), dup, missing, extra, 10000+acked.Load(), acked.Load()) + + hkv := must(hjs.KeyValue(ctx, "cfg")) + lk := must(kv.Keys(ctx)) + hk := must(hkv.Keys(ctx)) + diff := 0 + for _, k := range lk { + a, err1 := kv.Get(ctx, k) + b, err2 := hkv.Get(ctx, k) + if err1 != nil || err2 != nil || string(a.Value()) != string(b.Value()) { + diff++ + } + } + say("RESULT part4 KV cfg: keys legacy %d hub %d; keys whose latest value differs or is missing on hub: %d", len(lk), len(hk), diff) + + // cut-over: the hub stream stops sourcing and takes over the subjects + tc := time.Now() + subj := map[string][]string{"S": {"mesh.s.>"}, "KV_cfg": {"$KV.cfg.>"}} + for n, ss := range subj { + h := must(hjs.Stream(ctx, n)) + cfg := h.CachedInfo().Config + cfg.Sources = nil + cfg.Subjects = ss + if _, err := hjs.UpdateStream(ctx, cfg); err != nil { + say("cut-over %s FAILED: %v", n, err) + } + } + say("cut-over update (sources removed, subjects added, 2 streams): %s", time.Since(tc).Round(time.Millisecond)) + before := msgs(hjs, "S") + if _, err := hjs.Publish(ctx, "mesh.s.1", []byte("after-cutover")); err != nil { + say("RESULT part4 publish on hub after cut-over FAILED: %v", err) + } else { + say("RESULT part4 after cut-over a publish on the hub lands in the hub stream: %d -> %d messages (legacy stays %d)", before, msgs(hjs, "S"), msgs(ljs, "S")) + } + if _, err := hkv.Put(ctx, "after", []byte("1")); err != nil { + say("RESULT part4 KV put on hub after cut-over FAILED: %v", err) + } else { + g, err := hkv.Get(ctx, "after") + say("RESULT part4 KV put on hub after cut-over: readable %v", err == nil && string(g.Value()) == "1") + } +} diff --git a/test/bed-programs/signed-logins/gen.go b/test/bed-programs/signed-logins/gen.go new file mode 100644 index 0000000..007d88b --- /dev/null +++ b/test/bed-programs/signed-logins/gen.go @@ -0,0 +1,136 @@ +package main + +import ( + "encoding/json" + "flag" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/nats-io/jwt/v2" + "github.com/nats-io/nkeys" +) + +// Keys is what the driver on machine A needs to act as the operator, the identity seat and a system user. +type Keys struct { + Op, Sys, Acc, Sig string // seeds + A, B, C string // addresses +} + +func must[T any](v T, err error) T { + if err != nil { + panic(err) + } + return v +} + +func pubOf(kp nkeys.KeyPair) string { return must(kp.PublicKey()) } + +func fromSeed(s string) nkeys.KeyPair { return must(nkeys.FromSeed([]byte(s))) } + +func seedOf(kp nkeys.KeyPair) string { return string(must(kp.Seed())) } + +// accountJWT is the MESH account: JetStream limits, one signing key, a revocation list. +func accountJWT(op, acc, sig nkeys.KeyPair, revoked map[string]time.Time) string { + ac := jwt.NewAccountClaims(pubOf(acc)) + ac.Name = "MESH" + ac.Limits.JetStreamLimits = jwt.JetStreamLimits{MemoryStorage: 64 << 20, DiskStorage: 1 << 30, Streams: 20, Consumer: 50} + ac.SigningKeys.Add(pubOf(sig)) + if len(revoked) > 0 { + ac.Revocations = jwt.RevocationList{} + for k, t := range revoked { + ac.Revocations.Revoke(k, t) + } + } + return must(ac.Encode(op)) +} + +// userJWT: a user of the MESH account signed by the account's signing key, allowed everything. +func userJWT(acc, sig, ukp nkeys.KeyPair, name string) string { + uc := jwt.NewUserClaims(pubOf(ukp)) + uc.Name = name + uc.IssuerAccount = pubOf(acc) + uc.Pub.Allow.Add(">") + uc.Sub.Allow.Add(">") + return must(uc.Encode(sig)) +} + +func creds(token string, ukp nkeys.KeyPair) []byte { + return must(jwt.FormatUserConfig(token, []byte(seedOf(ukp)))) +} + +func gen(args []string) { + fs := flag.NewFlagSet("gen", flag.ExitOnError) + out := fs.String("out", "", "output directory") + a := fs.String("a", "", "address of A") + b := fs.String("b", "", "address of B") + c := fs.String("c", "", "address of C") + fs.Parse(args) + os.MkdirAll(*out, 0o755) + op := must(nkeys.CreateOperator()) + sys := must(nkeys.CreateAccount()) + acc := must(nkeys.CreateAccount()) + sig := must(nkeys.CreateAccount()) + oc := jwt.NewOperatorClaims(pubOf(op)) + oc.Name = "op" + oc.SystemAccount = pubOf(sys) + opJWT := must(oc.Encode(op)) + sc := jwt.NewAccountClaims(pubOf(sys)) + sc.Name = "SYS" + sysJWT := must(sc.Encode(op)) + accJ := accountJWT(op, acc, sig, nil) + + keys := Keys{seedOf(op), seedOf(sys), seedOf(acc), seedOf(sig), *a, *b, *c} + kj, _ := json.MarshalIndent(keys, "", " ") + w := func(name string, data []byte) { must(0, os.WriteFile(filepath.Join(*out, name), data, 0o600)) } + w("keys.json", kj) + w("op.jwt", []byte(opJWT)) + + // the legacy server joins as a leaf with one user in each of the two accounts + lk := must(nkeys.CreateUser()) + w("leaf-mesh.creds", creds(userJWT(acc, sig, lk, "legacy-leaf"), lk)) + sk := must(nkeys.CreateUser()) + suc := jwt.NewUserClaims(pubOf(sk)) + suc.Name = "legacy-leaf-sys" + w("leaf-sys.creds", creds(must(suc.Encode(sys)), sk)) + + common := func(name, domain string) string { + return fmt.Sprintf(`listen: 0.0.0.0:4222 +server_name: %s +http_port: 8222 +ping_interval: "2s" +ping_max: 2 +log_file: "/var/log/q8-%s.log" +jetstream: {store_dir: "/var/lib/q8/%s/js", domain: %s} +`, name, name, name, domain) + } + hub := func(name, peer string) string { + return common(name, "hub") + fmt.Sprintf(`operator: "/root/q8/op.jwt" +system_account: %s +resolver: {type: full, dir: "/var/lib/q8/%s/accts", allow_delete: false} +resolver_preload: {%s: %q, %s: %q} +cluster { name: HUB, listen: 0.0.0.0:6222, routes: ["nats-route://%s:6222"] } +leafnodes { listen: 0.0.0.0:7422 } +`, pubOf(sys), name, pubOf(sys), sysJWT, pubOf(acc), accJ, peer) + } + w("B.conf", []byte(hub("B", *c))) + w("C.conf", []byte(hub("C", *b))) + w("A.conf", []byte(common("A", "legacy")+fmt.Sprintf(`accounts { + MESH { jetstream: enabled + users: [ {user: w, password: pw} ] } + SYS { users: [ {user: sys, password: sys} ] } +} +system_account: SYS +leafnodes { remotes [ + {url: "nats-leaf://%s:7422", account: MESH, credentials: "/root/q8/leaf-mesh.creds"}, + {url: "nats-leaf://%s:7422", account: SYS, credentials: "/root/q8/leaf-sys.creds"} +] } +`, *b, *b))) +} + +func userClaimsFor(sys, ukp nkeys.KeyPair) string { + uc := jwt.NewUserClaims(pubOf(ukp)) + uc.Name = "sys" + return must(uc.Encode(sys)) +} diff --git a/test/bed-programs/signed-logins/go.mod b/test/bed-programs/signed-logins/go.mod new file mode 100644 index 0000000..fecdbe3 --- /dev/null +++ b/test/bed-programs/signed-logins/go.mod @@ -0,0 +1,21 @@ +module signedlogins + +go 1.26.0 + +require ( + github.com/nats-io/jwt/v2 v2.8.1 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.54.0 + github.com/nats-io/nkeys v0.4.16 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/test/bed-programs/signed-logins/go.sum b/test/bed-programs/signed-logins/go.sum new file mode 100644 index 0000000..98dc72c --- /dev/null +++ b/test/bed-programs/signed-logins/go.sum @@ -0,0 +1,25 @@ +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/test/bed-programs/signed-logins/main.go b/test/bed-programs/signed-logins/main.go new file mode 100644 index 0000000..e16adb7 --- /dev/null +++ b/test/bed-programs/signed-logins/main.go @@ -0,0 +1,36 @@ +// Command q8lab is the program behind test/integration/signed-logins.test.ts (research 053, question 8). +// +// q8lab gen -out DIR -a IP -b IP -c IP writes keys and the three servers' configs (runs on the workstation) +// q8lab server -conf FILE runs one nats-server 2.11.17 (embedded) until killed +// q8lab agent -peer IP on machine C: an HTTP switch that cuts and restores traffic to the peer +// q8lab wait -url URL waits until the URL answers 200 +// q8lab drive -keys FILE on machine A: runs the measurements and prints them +package main + +import ( + "fmt" + "os" +) + +func main() { + if len(os.Args) < 2 { + fmt.Println("usage: q8lab gen|server|agent|wait|drive") + os.Exit(2) + } + args := os.Args[2:] + switch os.Args[1] { + case "gen": + gen(args) + case "server": + runServer(args) + case "agent": + runAgent(args) + case "wait": + runWait(args) + case "drive": + drive(args) + default: + fmt.Println("unknown command", os.Args[1]) + os.Exit(2) + } +} diff --git a/test/bed-programs/signed-logins/serve.go b/test/bed-programs/signed-logins/serve.go new file mode 100644 index 0000000..ef2b190 --- /dev/null +++ b/test/bed-programs/signed-logins/serve.go @@ -0,0 +1,95 @@ +package main + +import ( + "flag" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/nats-io/nats-server/v2/server" +) + +func runServer(args []string) { + fs := flag.NewFlagSet("server", flag.ExitOnError) + conf := fs.String("conf", "", "config file") + fs.Parse(args) + opts, err := server.ProcessConfigFile(*conf) + if err != nil { + fmt.Println("config:", err) + os.Exit(1) + } + opts.NoSigs = true + opts.Debug = os.Getenv("Q8DEBUG") != "" + s, err := server.NewServer(opts) + if err != nil { + fmt.Println("server:", err) + os.Exit(1) + } + s.ConfigureLogger() + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + fmt.Println("not ready") + os.Exit(1) + } + fmt.Println("ready", server.VERSION) + ch := make(chan os.Signal, 1) + signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) + <-ch + s.Shutdown() +} + +// runAgent is a switch for machine C: it drops every packet to and from the peer, and takes the rule away again. +func runAgent(args []string) { + fs := flag.NewFlagSet("agent", flag.ExitOnError) + peer := fs.String("peer", "", "address to cut off") + fs.Parse(args) + nft := func(script string) string { + cmd := exec.Command("nft", "-f", "-") + cmd.Stdin = strings.NewReader(script) + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Sprintf("error: %v: %s", err, out) + } + return "ok" + } + http.HandleFunc("/block", func(w http.ResponseWriter, r *http.Request) { + nft("delete table inet q8cut\n") // ignore: it may not exist + io.WriteString(w, nft(fmt.Sprintf(`add table inet q8cut +add chain inet q8cut in { type filter hook input priority -10; policy accept; } +add chain inet q8cut out { type filter hook output priority -10; policy accept; } +add rule inet q8cut in ip saddr %s drop +add rule inet q8cut out ip daddr %s drop +`, *peer, *peer))) + }) + http.HandleFunc("/unblock", func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, nft("delete table inet q8cut\n")) + }) + fmt.Println("agent listening on :9099 for peer", *peer) + fmt.Println(http.ListenAndServe(":9099", nil)) +} + +func runWait(args []string) { + fs := flag.NewFlagSet("wait", flag.ExitOnError) + url := fs.String("url", "", "url") + secs := fs.Int("s", 60, "seconds") + fs.Parse(args) + end := time.Now().Add(time.Duration(*secs) * time.Second) + for time.Now().Before(end) { + if r, err := http.Get(*url); err == nil { + r.Body.Close() + if r.StatusCode == 200 { + fmt.Println("up") + return + } + } + time.Sleep(500 * time.Millisecond) + } + fmt.Println("timeout") + os.Exit(1) +} diff --git a/test/integration/signed-logins.test.ts b/test/integration/signed-logins.test.ts new file mode 100644 index 0000000..94f6f07 --- /dev/null +++ b/test/integration/signed-logins.test.ts @@ -0,0 +1,108 @@ +/** + * Research 053, question 8: the bus on signed logins (NATS operator mode), over a real network. + * + * Three machines on one segment: `a` is today's style server (static users, an account with + * JetStream), `b` and `c` are a two-server cluster in operator mode with a full resolver, and `a` is + * joined to `b` as a leaf node, the system account bridged. The measurements run in + * test/bed-programs/signed-logins (one Go program; it embeds nats-server 2.11.17, the mesh's + * version, so nothing is downloaded into the machines). It runs on `a` and reaches `b` and `c` + * across the segment: + * + * part 2 revocation: time for b and c to refuse a revoked user and close its open connection + * part 4 moving a stream and a KV bucket from a to b over the leaf link, with a writer running + * part 3 c cut off from b by a packet filter on c while a user is revoked, then restored: + * how long c still accepts the user with the periodic sync alone, and with a re-push + * triggered by the first system-account event that shows c is back + * + * The test asserts only that the measurements ran to the end; the numbers are the result and are + * printed. It proves nothing about the mesh's own software, which it does not use. + */ + +import { test, before, after } from "node:test"; +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { loadScenario } from "../../src/declaration/parse.ts"; +import { raise } from "../../src/lifecycle/raise.ts"; +import { destroy, exec, push } from "../../src/lifecycle/operate.ts"; +import { labIsUsable } from "./harness.ts"; + +const capability = await labIsUsable(); +const skip = capability.usable ? false : `lab not usable: ${capability.why}`; + +const ADDR = { a: "192.0.2.11", b: "192.0.2.12", c: "192.0.2.13" }; +let instanceId = ""; + +before(async () => { + if (skip) return; + instanceId = (await raise(loadScenario("scenarios/signed-logins.yml"), {})).instanceId; +}); + +after(async () => { + if (instanceId && !process.env["MESH_LAB_KEEP"]) await destroy(instanceId); +}); + +const sh = (machine: string, script: string, timeout = 120_000) => + exec(instanceId, machine, ["sh", "-c", script], timeout); + +test("research 053 q8: signed-login bus over a real network", { skip, timeout: 3_000_000 }, async () => { + const work = mkdtempSync(join(tmpdir(), "q8lab-")); + const bin = join(work, "q8lab"); + execFileSync("go", ["build", "-o", bin, "."], { + cwd: "test/bed-programs/signed-logins", + env: { ...process.env, CGO_ENABLED: "0", GOOS: "linux", GOARCH: "amd64", GOFLAGS: "-mod=readonly" }, + stdio: "inherit", + }); + const conf = join(work, "conf"); + mkdirSync(conf); + execFileSync(bin, ["gen", "-out", conf, "-a", ADDR.a, "-b", ADDR.b, "-c", ADDR.c], { stdio: "inherit" }); + + for (const m of ["a", "b", "c"]) { + await sh(m, "mkdir -p /root/q8 /var/lib/q8"); + await push(instanceId, m, bin, "/root/q8/q8lab", "0755"); + } + const files: Record = { + a: ["A.conf", "leaf-mesh.creds", "leaf-sys.creds", "keys.json"], + b: ["B.conf", "op.jwt"], + c: ["C.conf", "op.jwt"], + }; + for (const [m, list] of Object.entries(files)) { + for (const f of list) await push(instanceId, m, join(conf, f), `/root/q8/${f}`, "0600"); + } + + const start = (m: string, cmd: string, out: string) => + sh(m, `setsid nohup ${cmd} >/root/q8/${out} 2>&1 /root/q8/drive.log 2>&1; echo finished > /root/q8/drive.done' >/dev/null 2>&1 setTimeout(r, 5000)); + const chunk = (await sh("a", `tail -c +${offset + 1} /root/q8/drive.log`)).stdout; + offset += Buffer.byteLength(chunk); + finished = (await sh("a", "test -f /root/q8/drive.done && echo yes; true")).stdout.includes("yes"); + if (chunk.trim()) { console.log(chunk.trimEnd()); all += chunk; } + } + for (const m of ["a", "b", "c"]) { + const up = m.toUpperCase(); + const r = await sh(m, `grep -c '\\[ERR\\]' /var/log/q8-${up}.log; grep '\\[ERR\\]' /var/log/q8-${up}.log | cut -c1-220 | sort | uniq -c | sort -rn | head -8; true`); + console.log(`--- server ${up} log, error lines:\n${r.stdout.trim()}`); + } + writeFileSync(join(work, "drive.log"), all); + assert.ok(finished, "the measurements did not finish"); + assert.match(all, /DRIVE-DONE/); + assert.doesNotMatch(all, /panic:/); +});