A composition is applied only when the controller or the operator signed it; the node agent's integration tests on an embedded bus (R6-A A9, A1, A5)

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
 }

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")

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")

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 {

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")
 	}

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")

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"`) {

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")

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{}

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()

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):