Unify trunk on main: initialization → main #3
+65
-4
@@ -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()}
|
||||
}
|
||||
|
||||
@@ -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"}}
|
||||
})
|
||||
|
||||
+71
-7
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user