diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 611c74b..afa82ad 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -3,9 +3,13 @@ package main import ( "context" "crypto/ed25519" + "encoding/base64" "encoding/json" "fmt" + "os" + "path/filepath" "sort" + "strings" "sync" "time" @@ -179,6 +183,40 @@ func resourcesOf(cs []delivery.Composed) []declaration.Resource { return out } +// operatorKeysFile is the file beside the node's state that holds the operator's public keys, one ed25519 key in base64 a +// line (a line that starts with # is a note). The switch's last declaration writes it as a file resource, beside the +// controller's key the node was enrolled with (novox/hq ADR 0539 §3, §11); a node adopted later has it at its adoption. +const operatorKeysFile = "operator-keys" + +// trustedKeys are the keys a composition may be signed by: the controller's, which the node was enrolled with, and the +// operator's, read from operatorKeysFile each time it is asked, so a key that arrives while the agent runs is trusted at +// the next composition. A file that cannot be read trusts the controller's alone, and says so. +func trustedKeys(opts options, controller ed25519.PublicKey) func() []ed25519.PublicKey { + return func() []ed25519.PublicKey { + keys := []ed25519.PublicKey{} + if len(controller) == ed25519.PublicKeySize { + keys = append(keys, controller) + } + raw, err := os.ReadFile(filepath.Join(filepath.Dir(opts.state), operatorKeysFile)) + if err != nil { + return keys + } + for _, line := range strings.Split(string(raw), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + key, err := base64.StdEncoding.DecodeString(line) + if err != nil || len(key) != ed25519.PublicKeySize { + fmt.Fprintf(os.Stderr, "mesh-host: a line of %s is not an ed25519 public key in base64, and is not trusted\n", operatorKeysFile) + continue + } + keys = append(keys, ed25519.PublicKey(key)) + } + return keys + } +} + // keptByModule is the resources of the declaration the node kept before the switch, by the module they say; the node's // own are those that say none (ADR 0539 §9). func keptByModule(opts options, signer ed25519.PublicKey) func(module string) ([]json.RawMessage, bool) { @@ -240,6 +278,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess Composed: bucketWatch{session, composedBucket}, Haves: sessionHaves{session}, Kept: keptByModule(opts, signer), + Trusted: trustedKeys(opts, signer), Backup: backupNode(session), Say: say, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go new file mode 100644 index 0000000..df7f1d9 --- /dev/null +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -0,0 +1,325 @@ +//go:build integration + +package main + +import ( + "context" + "crypto/ed25519" + "encoding/base64" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/system" +) + +// The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition +// bucket and the have bucket, a session on the bus, and the real apply of a module's composition into a directory of the +// test. R6-A AC1, AC5 and AC9 (novox/hq ADR 0539). + +const itNode = "it1" + +type itBed struct { + t *testing.T + conn *nats.Conn + js nats.JetStreamContext + wants nats.KeyValue + composed nats.KeyValue + haves nats.KeyValue + dir string + opts options + session *link.Session + controll ed25519.PrivateKey + operator ed25519.PrivateKey + said []string + mu sync.Mutex +} + +func aServer(t *testing.T) *nats.Conn { + t.Helper() + srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the embedded server did not start") + } + t.Cleanup(srv.Shutdown) + conn, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(conn.Close) + return conn +} + +func newItBed(t *testing.T) *itBed { + t.Helper() + built, err := system.For("arch") + if err != nil || built.Confirm(context.Background(), apply.ExecRunner) != nil { + t.Skip("this machine is not one these tests can apply on") + } + was := builtFor + builtFor = "arch" + t.Cleanup(func() { builtFor = was }) + + b := &itBed{t: t, conn: aServer(t), dir: t.TempDir()} + if b.js, err = b.conn.JetStream(); err != nil { + t.Fatal(err) + } + composedBucket, _ := delivery.ComposedBucketFor(itNode) + for _, c := range []struct { + name string + into *nats.KeyValue + history uint8 + }{{delivery.AssignerWantsBucket, &b.wants, 8}, {composedBucket, &b.composed, delivery.ComposedHistory}, + {delivery.HaveBucket, &b.haves, delivery.HaveHistory}} { + kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: c.name, History: c.history}) + if err != nil { + t.Fatal(err) + } + *c.into = kv + } + b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second} + if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil { + t.Fatal(err) + } + _, b.controll, _ = ed25519.GenerateKey(nil) + _, b.operator, _ = ed25519.GenerateKey(nil) + return b +} + +// start runs the node agent under a session on the bus, trusting the controller's key, and the operator's when asked. +func (b *itBed) start(trustOperator bool) (stop func()) { + b.t.Helper() + if trustOperator { + pub := b.operator.Public().(ed25519.PublicKey) + if err := os.WriteFile(filepath.Join(filepath.Dir(b.opts.state), operatorKeysFile), + []byte("# the operator\n"+base64.StdEncoding.EncodeToString(pub)+"\n"), 0o600); err != nil { + b.t.Fatal(err) + } + } + b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, + Say: b.say} + b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + nodeAgent(ctx, b.opts, b.controll.Public().(ed25519.PublicKey), b.session, nil, b.say, func() {}, + &agentNow{}, func(context.Context, *agent.Agent) {}) + }() + return func() { cancel(); <-done } +} + +func (b *itBed) say(line string) { + b.mu.Lock() + b.said = append(b.said, line) + b.mu.Unlock() + b.t.Log(line) +} + +func fpIt(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } + +// compose writes a module's signed composition of one file into the node's bucket and returns its digest. +func (b *itBed) compose(module, fp, content string, data bool, signer ed25519.PrivateKey) string { + b.t.Helper() + path := filepath.Join(b.dir, module+".conf") + res, _ := json.Marshal(map[string]any{"id": module + ".conf", "type": "file", "path": path, "content": content, "module": module}) + c := delivery.Composed{Node: itNode, Module: module, FP: fp, Data: data, Resources: []json.RawMessage{res}} + signed, err := delivery.SignComposed(c, signer) + if err != nil { + b.t.Fatal(err) + } + raw, err := delivery.EncodeComposed(signed) + if err != nil { + b.t.Fatal(err) + } + dig, _ := delivery.ComposedDigestOf(signed) + key, _ := delivery.ComposedKey(itNode, module, fp) + if _, err := b.composed.Put(key, raw); err != nil { + b.t.Fatal(err) + } + return dig +} + +func (b *itBed) want(module, fp, digest, backup string) { + b.t.Helper() + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest, Backup: backup}) + if err != nil { + b.t.Fatal(err) + } + key, _ := delivery.WantKey(itNode, module) + if _, err := b.wants.Put(key, raw); err != nil { + b.t.Fatal(err) + } +} + +func (b *itBed) file(module string) string { + got, _ := os.ReadFile(filepath.Join(b.dir, module+".conf")) + return string(got) +} + +func (b *itBed) have(module string) delivery.Have { + key, _ := delivery.HaveKey(itNode, module) + e, err := b.haves.Get(key) + if err != nil { + return delivery.Have{} + } + h, _ := delivery.DecodeHave(e.Value()) + return h +} + +func within(t *testing.T, d time.Duration, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for !ok() { + if time.Now().After(deadline) { + t.Fatal(what) + } + time.Sleep(10 * time.Millisecond) + } +} + +// R6-A AC1: given a running agent on a bus, when a want naming a digest and that composition are written, the module is +// applied within one second and its have names the fingerprint and the digest; a composition whose digest no want names +// is not applied. +func TestNatsAWantAndItsCompositionAreAppliedWithinASecond(t *testing.T) { + b := newItBed(t) + stop := b.start(false) + defer stop() + time.Sleep(300 * time.Millisecond) // the watches are up + + fp := fpIt("a") + dig := b.compose("web", fp, "one\n", false, b.controll) + time.Sleep(300 * time.Millisecond) + if b.file("web") != "" { + t.Fatal("a composition no want names was applied") + } + began := time.Now() + b.want("web", fp, dig, "") + within(t, time.Second, "the module was not applied within one second of its want", func() bool { return b.file("web") == "one\n" }) + t.Logf("applied %s after the want", time.Since(began)) + within(t, 2*time.Second, "the have does not name the fingerprint and the digest", func() bool { + h := b.have("web") + return h.FP == fp && h.Composed == dig && h.Phase == delivery.PhaseRunning + }) + + // A want naming a digest the composition at its fingerprint does not have applies nothing. + fp2 := fpIt("b") + b.compose("web", fp2, "two\n", false, b.controll) + b.want("web", fp2, "sha256:"+strings.Repeat("0", 64), "") + time.Sleep(300 * time.Millisecond) + if b.file("web") != "one\n" { + t.Fatalf("a want naming another digest changed the module: %q", b.file("web")) + } +} + +// R6-A AC5: a data module that ran before asks its backup before any resource changes; a backup that fails holds the +// module (nothing applied, the have says held with the backup's words); when its want then says skip, it is applied. +func TestNatsADataModuleAsksItsBackupFirstAndIsHeldWhenItFails(t *testing.T) { + b := newItBed(t) + backupPollEvery = 20 * time.Millisecond + var mu sync.Mutex + var asked []string + var onDiskWhenAsked string + var nightOK bool + var at time.Time + if _, err := b.conn.Subscribe("mesh.seat.node-backup.tool.now."+itNode, func(m *nats.Msg) { + mu.Lock() + asked = append(asked, "now") + onDiskWhenAsked = b.file("db") + at = time.Now() + mu.Unlock() + _ = m.Respond([]byte(`{"result":{"started":"db"}}`)) + }); err != nil { + t.Fatal(err) + } + if _, err := b.conn.Subscribe("mesh.seat.node-backup.tool.backed-up."+itNode, func(m *nats.Msg) { + mu.Lock() + defer mu.Unlock() + if at.IsZero() { + _ = m.Respond([]byte(`{"result":[]}`)) + return + } + night, _ := json.Marshal(map[string]any{"module": "db", "lastNight": map[string]any{"ok": nightOK, + "at": time.Now().UTC(), "error": "restic: the repository is locked"}}) + _ = m.Respond([]byte(`{"result":[` + string(night) + `]}`)) + }); err != nil { + t.Fatal(err) + } + stop := b.start(false) + defer stop() + time.Sleep(300 * time.Millisecond) + + fp1, fp2 := fpIt("a"), fpIt("b") + b.want("db", fp1, b.compose("db", fp1, "v1\n", true, b.controll), "") + within(t, 3*time.Second, "the first composition was not applied", func() bool { return b.file("db") == "v1\n" }) + mu.Lock() + if len(asked) != 0 { + t.Fatalf("a module that never ran was backed up: %v", asked) + } + mu.Unlock() + + dig2 := b.compose("db", fp2, "v2\n", true, b.controll) + b.want("db", fp2, dig2, "") + within(t, 3*time.Second, "the module was not held behind the failed backup", func() bool { + h := b.have("db") + return h.Phase == delivery.PhaseHeld && strings.Contains(h.Detail, "repository is locked") + }) + mu.Lock() + if len(asked) != 1 || onDiskWhenAsked != "v1\n" { + t.Fatalf("the backup was asked %v, with %q on disk: it is asked before any resource changes", asked, onDiskWhenAsked) + } + mu.Unlock() + if b.file("db") != "v1\n" { + t.Fatalf("something was applied behind a failed backup: %q", b.file("db")) + } + if h := b.have("db"); h.FP != fp1 { + t.Fatalf("a held module should still name what it runs: %+v", h) + } + + b.want("db", fp2, dig2, delivery.BackupSkip) + within(t, 3*time.Second, "the module was not applied once its want said skip", func() bool { return b.file("db") == "v2\n" }) + within(t, 2*time.Second, "the have does not say running at the new fingerprint", func() bool { + h := b.have("db") + return h.Phase == delivery.PhaseRunning && h.FP == fp2 && h.Composed == dig2 + }) +} + +// R6-A AC9, the hand path end to end: with no controller running, a composition signed by the operator's key and a want +// naming it are applied and the have written; the same signed by a key nobody trusts is not. +func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) { + b := newItBed(t) + stop := b.start(true) + defer stop() + time.Sleep(300 * time.Millisecond) + + _, stranger, _ := ed25519.GenerateKey(nil) + fpS := fpIt("c") + b.want("evil", fpS, b.compose("evil", fpS, "no\n", false, stranger), "") + time.Sleep(300 * time.Millisecond) + if b.file("evil") != "" { + t.Fatal("a composition signed by a key nobody trusts was applied") + } + + fp := fpIt("d") + dig := b.compose("rescue", fp, "back\n", false, b.operator) + b.want("rescue", fp, dig, "") + within(t, 2*time.Second, "the operator's composition was not applied", func() bool { return b.file("rescue") == "back\n" }) + within(t, 2*time.Second, "the have was not written", func() bool { + h := b.have("rescue") + return h.FP == fp && h.Composed == dig && h.Phase == delivery.PhaseRunning + }) +} diff --git a/go.mod b/go.mod index c826857..9830c6e 100644 --- a/go.mod +++ b/go.mod @@ -3,14 +3,20 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024142924-d6cef5829bb7 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024143642-de5b33843f25 + github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 golang.org/x/sys v0.48.0 ) 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/jwt/v2 v2.8.1 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/go.sum b/go.sum index 994b40d..f5022eb 100644 --- a/go.sum +++ b/go.sum @@ -1,9 +1,17 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75 h1:AcYfyJpYaUufhI93VG76+b001msXY4XX7kuRjnf33HY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024142924-d6cef5829bb7 h1:tIzslR6sjUtCMlh1510t/D+dKXKqQUXp9dwTayhfJcs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024142924-d6cef5829bb7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024143642-de5b33843f25 h1:jxqWvEKoqeesxy2pgTEIKULNIFcLD7jmI6ZKTEfEnDw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024143642-de5b33843f25/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +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= @@ -12,5 +20,8 @@ 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/internal/agent/agent.go b/internal/agent/agent.go index 12e8375..9defd4c 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -12,6 +12,7 @@ package agent import ( "bytes" "context" + "crypto/ed25519" "encoding/json" "fmt" "sort" @@ -87,10 +88,11 @@ type Env struct { // Kept is the resources the node last applied for a module from the declaration it kept before the switch (ADR 0539 // §9), as JSON, and whether it held the module. Nil when the node kept none. Kept func(module string) ([]json.RawMessage, bool) - // Verify judges a composition's bytes before anything of it is read: a composition whose signer is not trusted is - // refused (ADR 0539 §3). Nil trusts the bucket, which only the controller and the operator can write. - Verify func(key string, raw []byte) error - Say func(string) + // Trusted is the public keys a composition may be signed by: the controller's and the operator's (ADR 0539 §3). A + // composition unsigned, signed by another key, or changed after it was signed is refused and nothing of it is + // applied. Nil trusts no key, so nothing is applied. + Trusted func() []ed25519.PublicKey + Say func(string) } // Agent follows the wants. @@ -281,23 +283,24 @@ func (a *Agent) takeComposition(u Update) string { return module } now := compNow{Raw: u.Value, Rev: u.Revision} - if a.env.Verify != nil { - if err := a.env.Verify(u.Key, u.Value); err != nil { - now.Err = fmt.Errorf("refused: %w", err) - a.comps[u.Key] = now - a.sayOnce("verify:"+u.Key, fmt.Sprintf("the composition %s is refused and nothing of it is applied: %v", u.Key, err)) - return module - } - } c, err := delivery.DecodeComposed(u.Value) if err == nil { + var trusted []ed25519.PublicKey + if a.env.Trusted != nil { + trusted = a.env.Trusted() + } + // The signature before anything of the composition is used: its digest is what the want names. now.C = c - now.Dig, err = delivery.ComposedDigestOf(c) - } - now.Err = err - if err != nil { + now.Dig, err = delivery.VerifyComposed(c, trusted) + if err != nil { + now.C = delivery.Composed{} + a.sayOnce("verify:"+u.Key, fmt.Sprintf("the composition %s is refused and nothing of it is applied: %v", u.Key, err)) + err = fmt.Errorf("refused: %w", err) + } + } else { a.sayOnce("decode:"+u.Key, fmt.Sprintf("the composition %s cannot be read, and nothing of it is applied: %v", u.Key, err)) } + now.Err = err a.comps[u.Key] = now return module } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 555c583..048f457 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -2,6 +2,7 @@ package agent import ( "context" + "crypto/ed25519" "encoding/json" "strings" "sync" @@ -13,6 +14,19 @@ import ( const node = "n1" +// The keys of the tests: the controller's and the operator's are trusted, a stranger's is not. +var controllerPub, controllerKey = newKey() +var operatorPub, operatorKey = newKey() +var _, strangerKey = newKey() + +func newKey() (ed25519.PublicKey, ed25519.PrivateKey) { + pub, key, err := ed25519.GenerateKey(nil) + if err != nil { + panic(err) + } + return pub, key +} + func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } // A watcher the test writes to; it says Caught once the keys written before the watch began are out. @@ -134,7 +148,8 @@ func newBed(t *testing.T) *bed { r, ok := b.kept[m] return r, ok }, - Say: func(s string) { t.Log(s) }, + Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub, operatorPub} }, + Say: func(s string) { t.Log(s) }, } return b } @@ -166,10 +181,22 @@ func composition(t *testing.T, module, fp string, data bool, rs ...json.RawMessa if err != nil { t.Fatal(err) } - if raw, err = delivery.EncodeComposed(c); err != nil { + return signedBy(t, c, key, controllerKey) +} + +// signedBy is c signed with signer, as the writer's code signs it: its key, its bytes and its digest. +func signedBy(t *testing.T, c delivery.Composed, key string, signer ed25519.PrivateKey) (string, []byte, string) { + t.Helper() + signed, err := delivery.SignComposed(c, signer) + if err != nil { t.Fatal(err) } - if digest, err = delivery.ComposedDigestOf(c); err != nil { + raw, err := delivery.EncodeComposed(signed) + if err != nil { + t.Fatal(err) + } + digest, err := delivery.ComposedDigestOf(signed) + if err != nil { t.Fatal(err) } return key, raw, digest @@ -461,10 +488,62 @@ func TestTheNodesOwnCompositionComesBeforeAnyModules(t *testing.T) { } } -// A composition the verifier refuses is never applied (the hook of ADR 0539 §3). -func TestARefusedCompositionIsNeverApplied(t *testing.T) { +// R6-A AC9: a composition signed by either trusted key is applied; one signed by another key, one changed after it was +// signed and one unsigned are refused and nothing of them is applied (ADR 0539 §3). +func TestOnlyACompositionSignedByATrustedKeyIsApplied(t *testing.T) { b := newBed(t) - b.env.Verify = func(string, []byte) error { return errRefused } + stop := b.run() + defer stop() + + plain := func(module, fp, content string) delivery.Composed { + return delivery.Composed{Node: node, Module: module, FP: fp, Resources: []json.RawMessage{file(module, "x", content)}} + } + put := func(c delivery.Composed, signer ed25519.PrivateKey, tamper func([]byte) []byte) { + key, _ := delivery.ComposedKey(node, c.Module, c.FP) + var raw []byte + var dig string + if signer == nil { + raw, _ = delivery.EncodeComposed(c) + dig, _ = delivery.ComposedDigestOf(c) + } else { + _, raw, dig = signedBy(t, c, key, signer) + } + if tamper != nil { + raw = tamper(raw) + } + b.comps.put(key, raw) + wk, wraw := wantFor(t, c.Module, c.FP, dig, "") + b.wants.put(wk, wraw) + } + + // Refused: signed by a stranger, changed after signing, unsigned. + put(plain("stranger", fpOf("1"), "x"), strangerKey, nil) + put(plain("changed", fpOf("2"), "before"), controllerKey, func(raw []byte) []byte { + return []byte(strings.Replace(string(raw), "before", "after!", 1)) + }) + put(plain("unsigned", fpOf("3"), "x"), nil, nil) + time.Sleep(200 * time.Millisecond) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a composition that is not signed by a trusted key was applied: %v", got) + } + + // Applied: signed by the controller, signed by the operator. + put(plain("bycontroller", fpOf("4"), "x"), controllerKey, nil) + put(plain("byoperator", fpOf("5"), "x"), operatorKey, nil) + eventually(t, "the trusted signers' compositions were not applied", func() bool { return len(b.appliedList()) == 2 }) + got := b.appliedList() + if !(got[0] == "bycontroller" && got[1] == "byoperator") && !(got[0] == "byoperator" && got[1] == "bycontroller") { + t.Fatalf("applied %v", got) + } + if h := b.haves.last("byoperator"); h.Phase != string(delivery.PhaseRunning) || h.FP != fpOf("5") { + t.Fatalf("the operator's module has no running have: %+v", h) + } +} + +// With no key trusted, nothing is applied: a node that does not know its signers fails closed. +func TestWithNoTrustedKeyNothingIsApplied(t *testing.T) { + b := newBed(t) + b.env.Trusted = nil stop := b.run() defer stop() fp := fpOf("a") @@ -472,14 +551,8 @@ func TestARefusedCompositionIsNeverApplied(t *testing.T) { b.comps.put(ck, craw) wk, wraw := wantFor(t, "web", fp, dig, "") b.wants.put(wk, wraw) - time.Sleep(150 * time.Millisecond) + time.Sleep(200 * time.Millisecond) if got := b.appliedList(); len(got) != 0 { - t.Fatalf("a refused composition was applied: %v", got) + t.Fatalf("applied with no trusted key: %v", got) } } - -type refusal string - -func (r refusal) Error() string { return string(r) } - -var errRefused error = refusal("signed by a key this node does not trust") diff --git a/internal/link/asktool_nats_test.go b/internal/link/asktool_nats_test.go index c1a4bcc..7e266cf 100644 --- a/internal/link/asktool_nats_test.go +++ b/internal/link/asktool_nats_test.go @@ -26,7 +26,7 @@ func TestNatsAToolCheckIsAskedOnTheLink(t *testing.T) { } q := &Session{Membership: Membership{Node: node}} - q.attach(t.Context(), &natsLink{conn: conn, js: js, node: node}) + q.Attach(t.Context(), &natsLink{conn: conn, js: js, node: node}) ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) defer cancel() a, err := q.AskTool(ctx, "sensors", "sensors_health") diff --git a/internal/link/have_nats_integration_test.go b/internal/link/have_nats_integration_test.go index cc7a6ff..eaa862c 100644 --- a/internal/link/have_nats_integration_test.go +++ b/internal/link/have_nats_integration_test.go @@ -43,7 +43,7 @@ func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) { Windows: []Window{{Step: "backup", Holds: []string{"db"}}}} } q := &Session{Membership: m, Timeout: 2 * time.Second} - q.attach(context.Background(), OverNATS{Conn: conn, JS: js}) + q.Attach(context.Background(), OverNATS{Conn: conn, JS: js}) ctx := context.Background() revisions := func() map[string]uint64 { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 2e6942d..54be3ea 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -139,7 +139,7 @@ func aHaveQueue(l *haveLink) (*Session, *[]string) { said = append(said, s) mu.Unlock() }} - q.attach(context.Background(), l) + q.Attach(context.Background(), l) return q, &said } @@ -277,7 +277,7 @@ func TestAFailedHaveWriteIsSaidAndNeverBlocksTheReport(t *testing.T) { func TestNoBucketAndNoFactsWriteNothing(t *testing.T) { plain := newQuietLink() q := &Session{Membership: Membership{Node: "n1"}, Timeout: time.Second} - q.attach(context.Background(), plain) + q.Attach(context.Background(), plain) if !q.SayFacts(context.Background(), appliedModules()) { t.Fatal("the report did not go out") } diff --git a/internal/link/health_test.go b/internal/link/health_test.go index 5ef80ef..db7935a 100644 --- a/internal/link/health_test.go +++ b/internal/link/health_test.go @@ -30,7 +30,7 @@ func TestAHealthStatementIsSaidOnTheLinkOpenNow(t *testing.T) { t.Fatal("said with no link open") } l := &sayingLink{quietLink: newQuietLink()} - q.attach(t.Context(), l) + q.Attach(t.Context(), l) if !q.SayHealth(t.Context(), h) || len(l.said) != 1 { t.Fatalf("not said on the open link: %d", len(l.said)) } @@ -42,8 +42,8 @@ func TestAHealthStatementIsSaidOnTheLinkOpenNow(t *testing.T) { t.Fatalf("said %+v", got) } // A link that cannot say it is not an error: the next report carries the statement. - q.detach(l) - q.attach(t.Context(), newQuietLink()) + q.Detach(l) + q.Attach(t.Context(), newQuietLink()) if q.SayHealth(t.Context(), h) { t.Fatal("said on a link that cannot") } @@ -83,16 +83,16 @@ func TestAToolCheckIsAskedOnTheLinkTheEngineHolds(t *testing.T) { } quiet := newQuietLink() - q.attach(t.Context(), quiet) + q.Attach(t.Context(), quiet) if _, err := q.AskTool(t.Context(), "sensors", "sensors_health"); err == nil || err.Error() != "the link to the bus open now cannot ask a tool" { t.Fatalf("on a link that cannot ask: %v", err) } - q.detach(quiet) + q.Detach(quiet) // The link Open returns, with no connection under it: the question must reach its Ask (and fail // there), never stop at the queue's assertion. held := &natsLink{node: "anchor"} - q.attach(t.Context(), held) + q.Attach(t.Context(), held) _, err := q.AskTool(t.Context(), "sensors", "sensors_health") if err == nil { t.Fatal("asked on no connection and was answered") diff --git a/internal/link/movesecret_test.go b/internal/link/movesecret_test.go index 8457711..41ae611 100644 --- a/internal/link/movesecret_test.go +++ b/internal/link/movesecret_test.go @@ -113,7 +113,7 @@ func TestNatsAMoveIsAskedOfTheNodeAndTheVaultIsAskedInItsName(t *testing.T) { defer vault.Unsubscribe() l := &natsLink{conn: conn, js: js, node: m.Node, signer: m.Signer} q := &Session{Membership: m} - q.attach(t.Context(), l) + q.Attach(t.Context(), l) l.HearMoveSecrets(func(s MoveSecret) (MovedSecret, error) { raw, err := q.AskVault(t.Context(), "move-in", map[string]string{"module": s.Module}) if err != nil || !strings.Contains(string(raw), `"call":"c1"`) { diff --git a/internal/link/replay331_test.go b/internal/link/replay331_test.go index a603fd3..087900e 100644 --- a/internal/link/replay331_test.go +++ b/internal/link/replay331_test.go @@ -14,7 +14,7 @@ import ( // link and fail there, not stop at the queue as if no link were open. func TestReplay331(t *testing.T) { q := &Session{Membership: Membership{Node: "laptop"}} - q.attach(t.Context(), &natsLink{node: "laptop"}) + q.Attach(t.Context(), &natsLink{node: "laptop"}) _, err := q.AskTool(t.Context(), "sensors", "sensors_health") if err == nil { t.Fatal("asked on no connection and was answered") diff --git a/internal/link/run.go b/internal/link/run.go index b19a2a5..3d17258 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -175,8 +175,8 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann say("in the mesh, hearing what the controller asks of this node") // The session says this node's facts as the link is handed to it, and is not let go while it is mid-act. - queue.attach(ctx, link) - defer queue.detach(link) + queue.Attach(ctx, link) + defer queue.Detach(link) // And the mesh asking this node to say its facts again, on a link that hears the question. var asks <-chan struct{} diff --git a/internal/link/session.go b/internal/link/session.go index 541f773..b90f733 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -116,8 +116,8 @@ func (q *Session) FactsAsked(ctx context.Context) { } } -// attach is a link opening: it is the link facts are said on from now. -func (q *Session) attach(ctx context.Context, bus Bus) { +// Attach is a link opening: it is the link facts are said on from now. +func (q *Session) Attach(ctx context.Context, bus Bus) { q.init() q.mu.Lock() q.bus = bus @@ -140,8 +140,8 @@ func (q *Session) WithLink(ctx context.Context) context.Context { return WithFiles(ctx, bus) } -// detach is the link closing: nothing more is said on it. -func (q *Session) detach(bus Bus) { +// Detach is the link closing: nothing more is said on it. +func (q *Session) Detach(bus Bus) { q.init() q.saying.Lock() defer q.saying.Unlock() diff --git a/internal/link/session_test.go b/internal/link/session_test.go index cd421a2..909ed82 100644 --- a/internal/link/session_test.go +++ b/internal/link/session_test.go @@ -64,7 +64,7 @@ func TestTheFactsEventCarriesTheNodeEnginesWriterEpochAndSequence(t *testing.T) m, _ := aMember(t) l := newQuietLink() s := &Session{Membership: m, Timeout: time.Second} - s.attach(context.Background(), l) + s.Attach(context.Background(), l) if !s.SayFacts(context.Background(), Report{Firewall: "none"}) || !s.SayFacts(context.Background(), Report{}) { t.Fatal("the facts were not taken") @@ -85,7 +85,7 @@ func TestTheFactsEventCarriesTheNodeEnginesWriterEpochAndSequence(t *testing.T) // A later start orders after this one with nothing kept on disk. time.Sleep(2 * time.Millisecond) later := &Session{Membership: m, Timeout: time.Second} - later.attach(context.Background(), l) + later.Attach(context.Background(), l) later.SayFacts(context.Background(), Report{}) if last := l.said()[2]; last.Epoch <= got[1].Epoch { t.Fatalf("a later start has epoch %d, not above %d", last.Epoch, got[1].Epoch) @@ -100,8 +100,8 @@ func TestFactsWithNoLinkOpenAreNotSaid(t *testing.T) { if s.SayFacts(context.Background(), Report{}) { t.Fatal("facts were said with no link") } - s.attach(context.Background(), l) - s.detach(l) + s.Attach(context.Background(), l) + s.Detach(l) if s.Linked() || s.SayFacts(context.Background(), Report{}) || len(l.said()) != 0 { t.Fatal("facts were said on a link that was let go") } @@ -113,7 +113,7 @@ func TestALinkOpeningSaysTheFacts(t *testing.T) { l := newQuietLink() opened := make(chan struct{}, 1) s := &Session{Membership: m, Timeout: time.Second, OnLink: func(context.Context) { opened <- struct{}{} }} - s.attach(context.Background(), l) + s.Attach(context.Background(), l) select { case <-opened: case <-time.After(2 * time.Second):