diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index dd342f2..38249bb 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -546,16 +546,64 @@ func runLink(ctx context.Context, opts options) error { fmt.Printf("node %s, linking to %s\n", mine.Node, mine.Membership.Broker) - apply := func(ctx context.Context, raw []byte) link.Report { - return applyDeclared(ctx, opts, raw) + apply := func(ctx context.Context, raw, signature []byte) link.Report { + return applyAndKeep(ctx, opts, raw, &store.Declared{Declaration: raw, Signature: signature}) } - return link.Run(ctx, link.Membership{ + say := func(line string) { fmt.Println(line) } + + // Two things at once, and the second is what makes disconnection ordinary. The link brings + // new declarations; this holds the machine in the last one whether the link is up or not. A + // laptop shut for a week comes back and reconciles — it does not come back and ask what it is + // (novox/hq ADR 0004). + go holdTheMachine(ctx, opts, mine, say) + + return link.Hold(ctx, link.Membership{ Node: mine.Node, Broker: mine.Membership.Broker, Fingerprint: mine.Membership.Fingerprint, Password: mine.Membership.Password, Signer: mine.Membership.Signer, - }, apply, func(line string) { fmt.Println(line) }, opts.timeout) + }, apply, say, opts.timeout) +} + +// ReconcileEvery is how often a node re-applies what it was last told. +// +// Not driven by the link. Changes are pushed, so this is not polling for them — it is the answer +// to a machine drifting: a file edited by hand, a container stopped by somebody, a service that +// died. A node that only acted when told would hold its state exactly until something else +// changed it, and then for ever. +const ReconcileEvery = 5 * time.Minute + +func holdTheMachine(ctx context.Context, opts options, mine identity.Identity, say link.Announce) { + ticker := time.NewTicker(ReconcileEvery) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + + declared, err := store.LoadDeclared(store.DeclaredPath(opts.state), mine.Membership.Signer) + if errors.Is(err, store.ErrNothingDeclared) { + // Nothing to hold this machine to yet. Ordinary on a node that has enrolled and not + // been assigned anything. + continue + } + if err != nil { + say("cannot re-apply what this node was told: " + err.Error()) + continue + } + + report := applyDeclared(ctx, opts, declared) + switch { + case report.Refused != "": + say("what this node was last told no longer applies: " + report.Refused) + case len(report.Failed) > 0: + say(fmt.Sprintf("holding this machine: %d applied, and %v", len(report.Applied), report.Failed)) + } + } } // applyDeclared applies a declaration that has already been proved to come from the mesh. @@ -564,6 +612,12 @@ func runLink(ctx context.Context, opts options) error { // the question "is this from the mesh I joined" is settled — which is why this can treat the // bytes as instructions. func applyDeclared(ctx context.Context, opts options, raw []byte) link.Report { + return applyAndKeep(ctx, opts, raw, nil) +} + +// applyAndKeep applies a declaration and, when it came from the mesh, keeps it so this node can +// go on obeying it while disconnected. +func applyAndKeep(ctx context.Context, opts options, raw []byte, signed *store.Declared) link.Report { declared, err := declaration.Parse(raw) if err != nil { return link.Report{Refused: err.Error()} @@ -602,6 +656,13 @@ func applyDeclared(ctx context.Context, opts options, raw []byte) link.Report { for _, change := range outcome.Outcomes { report.Applied = append(report.Applied, change.ID) } + // Kept whichever way it went, so a node that is disconnected next minute still knows what it + // was told. Saved after applying rather than before: what is kept is what this node acted on. + if signed != nil { + if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil { + report.Failed = map[string]string{"keeping the declaration": err.Error()} + } + } if applyErr != nil { report.Failed = map[string]string{"apply": applyErr.Error()} } diff --git a/internal/link/messages_test.go b/internal/link/messages_test.go index 4be61eb..970a2aa 100644 --- a/internal/link/messages_test.go +++ b/internal/link/messages_test.go @@ -14,7 +14,7 @@ func verified(t *testing.T, signer ed25519.PublicKey, body []byte) (Report, bool t.Helper() applied := false report := handleBody(context.Background(), Membership{Node: "anchor", Signer: signer}, body, - func(context.Context, []byte) Report { + func(context.Context, []byte, []byte) Report { applied = true return Report{Applied: []string{"something"}} }) diff --git a/internal/link/run.go b/internal/link/run.go index db68252..882e8b6 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -29,18 +29,77 @@ type Membership struct { } // Applier is what the host does with a declaration that has been proved to come from the mesh. -type Applier func(ctx context.Context, declaration []byte) Report +// +// It receives the signature as well as the declaration, so the host can keep both: what it was +// told is kept signed and verified again when it is read back, which means the file on disk is +// trusted for the same reason the message was rather than for being local. +type Applier func(ctx context.Context, declaration, signature []byte) Report // Announce is how the link says what is happening, so a node running unattended leaves an // account of it. Nil is allowed and means say nothing. type Announce func(string) -// Run holds the link open, applying what arrives and reporting what happened. +// Hold keeps this node in the mesh, reconnecting for as long as it is asked to. // -// Outbound only, and nothing listens on this machine. The connection is the node's presence in -// the mesh: while it is up the node is enrolled, and while it is down the node is disconnected — -// which is an ordinary situation and not a failure, so this returns rather than panicking and -// leaves restarting to whatever supervises it. +// Disconnection is an ordinary situation and not a failure (novox/hq ADR 0004), so this does not +// give up. A laptop shut for a week comes back and reconnects; it does not come back needing +// somebody to start it again. +// +// The backoff exists because the two common reasons differ in how long they last: a broker +// restarting is back in seconds, and a machine that has moved to a network with no route may be +// hours. Retrying every second for hours is a node shouting into nothing; waiting a minute after +// a broker blip is a node that is needlessly late. So it starts fast and slows down, and resets +// once a connection has actually held. +func Hold(ctx context.Context, m Membership, apply Applier, say Announce, timeout time.Duration) error { + const ( + first = 2 * time.Second + most = 2 * time.Minute + // A connection that lasted this long counts as having worked, so the next failure starts + // from the bottom again. Without it a node that reconnects and immediately drops climbs + // to the maximum and stays there, long after whatever caused it went away. + settled = 30 * time.Second + ) + wait := first + + for { + began := time.Now() + err := Run(ctx, m, apply, say, timeout) + if ctx.Err() != nil { + return nil + } + if time.Since(began) > settled { + wait = first + } + + switch { + case errors.Is(err, ErrWrongCertificate): + // Said in full every time rather than folded into a retry count. This does not mean + // the network is down; it means what answered is not the mesh this node joined, and + // no amount of waiting fixes it. The node keeps running what it was last told, which + // is the right thing to do while somebody works out what happened. + say("the broker is not the one this node joined: " + err.Error()) + say("this will not fix itself. This node keeps running what it was last told.") + case err != nil: + say(fmt.Sprintf("disconnected: %v — trying again in %s", err, wait)) + default: + say(fmt.Sprintf("the link closed — trying again in %s", wait)) + } + + select { + case <-ctx.Done(): + return nil + case <-time.After(wait): + } + if wait *= 2; wait > most { + wait = most + } + } +} + +// Run holds the link open once, applying what arrives and reporting what happened. +// +// Outbound only, and nothing listens on this machine. Returns when the link ends, for any reason; +// Hold is what decides whether to open it again. func Run(ctx context.Context, m Membership, apply Applier, say Announce, timeout time.Duration) error { if say == nil { say = func(string) {} @@ -89,6 +148,11 @@ func Run(ctx context.Context, m Membership, apply Applier, say Announce, timeout if err != nil { return err } + // Said, because it is the event anybody watching actually wants. Without it a node logs + // every failure and nothing on success, so a log full of "trying again" and then silence + // reads as still broken when it means the opposite. + say("in the mesh, consuming " + queue) + closed := conn.NotifyClose(make(chan *amqp.Error, 1)) // Published mandatory, so the broker hands back anything it cannot route rather than @@ -150,7 +214,7 @@ func handleBody(ctx context.Context, m Membership, body []byte, apply Applier) R if !ed25519.Verify(m.Signer, signed.Declaration, signed.Signature) { return Report{Node: m.Node, Refused: ErrForged.Error()} } - return apply(ctx, signed.Declaration) + return apply(ctx, signed.Declaration, signed.Signature) } func publishReport(ctx context.Context, channel *amqp.Channel, m Membership, report Report, diff --git a/internal/store/declared.go b/internal/store/declared.go new file mode 100644 index 0000000..5d2e80d --- /dev/null +++ b/internal/store/declared.go @@ -0,0 +1,108 @@ +package store + +import ( + "crypto/ed25519" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" +) + +// What the mesh last told this node to be, kept so it can go on being it. +// +// novox/hq ADR 0004: a disconnected node keeps reconciling against its own store, so it holds its +// machine in the last state it was told. A laptop shut for a week comes back and reconciles; it +// does not come back and ask what it is. +// +// That needs the declaration itself. The record of what was *applied* is not enough to re-apply: +// it holds an id, a type and a target, which is what removal needs and not what creation needs. +// So the declaration is kept whole. +// +// **Kept signed, and verified again on every load.** The signature is not decoration here: this +// file is on a machine, and a node that read it back unverified would apply whatever was in it. +// Anyone able to write it already has root — but the check costs nothing, and it means the file +// is trusted for the same reason the message was, rather than for being local. + +// DeclaredName is where it lives, beside the state. +const DeclaredName = "declared.json" + +// DeclaredPath is where the last declaration lives, given where the state lives. +func DeclaredPath(statePath string) string { + return filepath.Join(filepath.Dir(statePath), DeclaredName) +} + +// Declared is the last thing the mesh said, and the signature it came with. +type Declared struct { + Declaration []byte `json:"declaration"` + Signature []byte `json:"signature"` +} + +// ErrNothingDeclared means the mesh has never told this node anything. +// +// An ordinary state, not a fault: a node that has enrolled and not yet been sent a declaration +// has nothing to reconcile against, and that is different from having lost it. +var ErrNothingDeclared = errors.New("the mesh has not told this node anything yet") + +// SaveDeclared keeps what the mesh said, so a disconnected node can go on obeying it. +func SaveDeclared(path string, d Declared) error { + if len(d.Declaration) == 0 { + return errors.New("refusing to keep an empty declaration") + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + raw, err := json.Marshal(d) + if err != nil { + return err + } + + tmp, err := os.CreateTemp(filepath.Dir(path), ".declared-*") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + if err := tmp.Chmod(0o600); err != nil { + tmp.Close() + return err + } + if _, err := tmp.Write(raw); err != nil { + tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmp.Name(), path) +} + +// LoadDeclared reads it back and proves it is still the mesh's. +// +// Verified against the signing key this node holds, which came from its token. A declaration on +// disk that does not verify is refused rather than applied: either the file was changed, or this +// node now believes a different mesh — and applying it either way would be applying something +// nobody in this mesh said. +func LoadDeclared(path string, signer ed25519.PublicKey) ([]byte, error) { + raw, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return nil, ErrNothingDeclared + } + if err != nil { + return nil, fmt.Errorf("this node was told something and cannot read it back: %w", err) + } + + var d Declared + if err := json.Unmarshal(raw, &d); err != nil { + return nil, fmt.Errorf("what this node was told is unreadable at %s: %w", path, err) + } + if !ed25519.Verify(signer, d.Declaration, d.Signature) { + return nil, fmt.Errorf( + "what this node kept at %s is not signed by the mesh it joined. It will not be "+ + "applied — either the file was changed, or this node's signing key was", path) + } + return d.Declaration, nil +} diff --git a/internal/store/declared_test.go b/internal/store/declared_test.go new file mode 100644 index 0000000..4d36a0f --- /dev/null +++ b/internal/store/declared_test.go @@ -0,0 +1,140 @@ +package store + +import ( + "crypto/ed25519" + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func signedBy(t *testing.T, body string) (ed25519.PublicKey, Declared) { + t.Helper() + public, private, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatal(err) + } + return public, Declared{ + Declaration: []byte(body), + Signature: ed25519.Sign(private, []byte(body)), + } +} + +func TestNothingDeclaredIsNotAFault(t *testing.T) { + // A node that has enrolled and not yet been assigned anything has nothing to hold its machine + // to, and that is an ordinary state — different from having lost what it was told. + public, _ := signedBy(t, "{}") + _, err := LoadDeclared(DeclaredPath(filepath.Join(t.TempDir(), "state.json")), public) + if !errors.Is(err, ErrNothingDeclared) { + t.Fatalf("a node that was never told anything gave %v", err) + } +} + +func TestWhatWasKeptIsWhatComesBack(t *testing.T) { + path := DeclaredPath(filepath.Join(t.TempDir(), "state.json")) + public, d := signedBy(t, `{"declaration":1,"resources":[]}`) + if err := SaveDeclared(path, d); err != nil { + t.Fatal(err) + } + + back, err := LoadDeclared(path, public) + if err != nil { + t.Fatal(err) + } + if string(back) != string(d.Declaration) { + t.Errorf("kept %q and read back %q", d.Declaration, back) + } +} + +func TestWhatWasKeptIsVerifiedAgainOnLoad(t *testing.T) { + // This file is on a machine, and a node reading it back unverified would apply whatever is in + // it. Anyone able to write it already has root — but the check costs nothing, and it means + // the file is trusted for the same reason the message was rather than for being local. + path := DeclaredPath(filepath.Join(t.TempDir(), "state.json")) + public, d := signedBy(t, `{"declaration":1,"resources":[]}`) + if err := SaveDeclared(path, d); err != nil { + t.Fatal(err) + } + + // Somebody edits it, keeping the signature. + tampered, err := json.Marshal(Declared{ + Declaration: []byte(`{"declaration":1,"resources":["something else"]}`), + Signature: d.Signature, + }) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, tampered, 0o600); err != nil { + t.Fatal(err) + } + + if _, err := LoadDeclared(path, public); err == nil { + t.Fatal("an edited declaration was read back and would have been applied") + } +} + +func TestAnotherMeshsDeclarationIsRefused(t *testing.T) { + // The same check answers a second question: this node now believes a different signing key, + // so what it kept is not this mesh's. Applying it would be applying something nobody in this + // mesh said. + path := DeclaredPath(filepath.Join(t.TempDir(), "state.json")) + _, d := signedBy(t, `{"declaration":1}`) + if err := SaveDeclared(path, d); err != nil { + t.Fatal(err) + } + other, _ := signedBy(t, "unrelated") + + _, err := LoadDeclared(path, other) + if err == nil { + t.Fatal("a declaration signed by another mesh was accepted") + } + if !strings.Contains(err.Error(), "not signed by the mesh it joined") { + t.Errorf("the refusal does not say what is wrong: %v", err) + } +} + +func TestWhatWasKeptIsNotWorldReadable(t *testing.T) { + path := DeclaredPath(filepath.Join(t.TempDir(), "state.json")) + _, d := signedBy(t, `{"declaration":1}`) + if err := SaveDeclared(path, d); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm()&0o077 != 0 { + t.Errorf("what this node was told is mode %04o", info.Mode().Perm()) + } +} + +func TestKeepingLeavesNoHalfWrittenFile(t *testing.T) { + dir := t.TempDir() + path := DeclaredPath(filepath.Join(dir, "state.json")) + _, d := signedBy(t, `{"declaration":1}`) + for i := 0; i < 3; i++ { + if err := SaveDeclared(path, d); err != nil { + t.Fatal(err) + } + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if strings.HasPrefix(e.Name(), ".declared-") { + t.Errorf("a temporary file survived: %s", e.Name()) + } + } +} + +func TestAnEmptyDeclarationIsNotKept(t *testing.T) { + // It would read back as an instruction to own nothing, and a node that acted on it would + // remove everything the mesh had given it. + path := DeclaredPath(filepath.Join(t.TempDir(), "state.json")) + if err := SaveDeclared(path, Declared{}); err == nil { + t.Fatal("an empty declaration was kept") + } +}