diff --git a/cmd/mesh-host/enrol_seat.go b/cmd/mesh-host/enrol_seat.go new file mode 100644 index 0000000..4d253a8 --- /dev/null +++ b/cmd/mesh-host/enrol_seat.go @@ -0,0 +1,93 @@ +package main + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "os" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/identity" + "github.com/novox/mesh-host/internal/link" +) + +// enrolThroughSeat joins this machine through the seat mesh-enrolment (novox/hq ADR 0587 §6, issue 774): it redeems the token +// with the keys it made, verifies the vault's signed answer against the key the token carries before it keeps anything, keeps +// its membership and its login, dials as the login node-logins names and confirms. handled is false when the bus does not serve +// the seat to this token (the controller still holds its lease, so its enrolment subject is the way in): nothing was sent or +// kept then, and the caller enrols as before. +func enrolThroughSeat(ctx context.Context, opts options, token identity.Token, node string, mine identity.Identity, + sealing identity.SealingKey, serving identity.ServingKey, reported map[string]any, found *link.Tunnel, identityPath string) (handled bool, err error) { + + req := delivery.Redeem{Node: node, PublicKey: base64.StdEncoding.EncodeToString(mine.Public), OverlayKey: mine.Overlay.Public, + SealingKey: sealing.Public, ServingKey: serving.Public, Profile: reported} + if found != nil { + if req.Tunnel, err = json.Marshal(found); err != nil { + return true, err + } + } + req.Proof = base64.StdEncoding.EncodeToString(mine.Sign(delivery.RedeemProof(node, req.PublicKey, req.OverlayKey, req.SealingKey, req.ServingKey))) + to := link.Approach{Address: token.Broker, Fingerprint: token.Fingerprint, Transport: link.OnNATS} + membership := link.Membership{Node: node, Broker: token.Broker, Fingerprint: token.Fingerprint, Transport: link.OnNATS} + j := link.Joining{Node: node, Signer: token.SignerKey(), + Redeem: func(ctx context.Context) (delivery.RedeemAnswer, error) { + return link.Redeem(ctx, to, node, token.Secret, req, opts.timeout) + }, + Open: func(a delivery.RedeemAnswer) (string, error) { + value, err := sealing.Open(a.Account) + return string(value), err + }, + Keep: func(a delivery.RedeemAnswer, password string) error { + // The mesh's name for this node wins over what the machine called itself, as on the controller's path. + mine.Node = a.Node + mine.Membership = identity.Membership{Broker: token.Broker, Fingerprint: token.Fingerprint, Signer: token.Signer, + Password: password, Transport: link.OnNATS} + if err := writeLogins(loginsFile(opts.state), link.LoginsState{Current: a.Login, Values: map[string]string{a.Login: password}}); err != nil { + return fmt.Errorf("cannot keep the login %s: %w", a.Login, err) + } + if err := identity.Save(identityPath, mine); err != nil { + return err + } + if !mine.Membership.Joined() { + return errors.New("the answer did not say how to reach the mesh again, so the identity could not be used after a restart") + } + return keepJoinedKeys(opts, mine, sealing, serving) + }, + Dial: func(ctx context.Context, login, password string) (link.Confirming, error) { + return link.DialConfirming(ctx, membership, login, password, opts.timeout) + }, + } + answer, err := j.Run(ctx) + switch { + case errors.Is(err, link.ErrNoSeat): + // The path before the lease write: the controller's enrolment subject is granted to the token only while the + // controller holds its lease, so after the write this fallback is refused too, and that refusal is what is said. + fmt.Printf("the seat mesh-enrolment is not served to this token (%v); enrolling through the controller's enrolment "+ + "subject, the path before the lease write, which the bus refuses once the lease is written\n", err) + return false, nil + case err != nil: + return true, err + } + fmt.Printf("\nenrolled as %s through the seat mesh-enrolment, dialled as %s and confirmed\n", answer.Node, answer.Login) + fmt.Printf(" identity %s\n", identityPath) + fmt.Println(" nothing is applied until the placement of this node names a module") + return true, nil +} + +// keepJoinedKeys writes the private halves of the keys this machine made, after its identity: an identity naming a key that is +// not there looks joined and cannot come up, so the identity is written first and a failure here is said plainly. +func keepJoinedKeys(opts options, mine identity.Identity, sealing identity.SealingKey, serving identity.ServingKey) error { + if err := os.WriteFile(identity.OverlayKeyPath(opts.state), []byte(mine.Overlay.Private+"\n"), 0o600); err != nil { + return fmt.Errorf("cannot write this node's overlay key: %w", err) + } + if err := os.WriteFile(identity.SealingKeyPath(opts.state), []byte(sealing.Private+"\n"), 0o600); err != nil { + return fmt.Errorf("cannot write this node's sealing key: %w", err) + } + if err := identity.WriteServingKey(identity.ServingKeyPath(opts.state), serving); err != nil { + return fmt.Errorf("cannot write this node's serving key: %w", err) + } + return nil +} diff --git a/cmd/mesh-host/logins.go b/cmd/mesh-host/logins.go new file mode 100644 index 0000000..ab29a6f --- /dev/null +++ b/cmd/mesh-host/logins.go @@ -0,0 +1,154 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/identity" + "github.com/novox/mesh-host/internal/link" +) + +// loginsFile is where this engine keeps the logins it holds between runs (novox/hq ADR 0583 §1): beside its identity, readable by +// its owner alone, as the identity's membership password is. +func loginsFile(state string) string { + return filepath.Join(filepath.Dir(identity.Path(state)), "logins.json") +} + +// loadLogins is the engine's logins: what it kept, with the membership's password as the first login's. +func loadLogins(state, node, first string) *link.Logins { + path := loginsFile(state) + var saved link.LoginsState + if raw, err := os.ReadFile(path); err == nil { + _ = json.Unmarshal(raw, &saved) + } + return link.NewLogins(node, first, saved, func(s link.LoginsState) error { return writeLogins(path, s) }) +} + +func writeLogins(path string, s link.LoginsState) error { + raw, err := json.Marshal(s) + if err != nil { + return err + } + tmp := path + ".new" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +// loginsFollowAgain bounds the wait before a watch of node-logins that could not be made is tried again on the same link: +// the vault may not serve rotation yet, and until it does this node is not granted the read. +var loginsFollowAgain = 2 * time.Minute + +// followLogins follows this node's key of the vault's node-logins on every link, as a watch (never a poll), for as long as ctx +// lasts, and hands each value to hear. A watch that cannot be made is said once and tried again. +func followLogins(ctx context.Context, session *link.Session, node string, hear func(delivery.NodeLogin), say link.Announce) { + said := "" + for ctx.Err() == nil { + linked, cancel, err := session.WhenLinked(ctx) + if err != nil { + return + } + watching, stop := context.WithCancel(linked) + updates, werr := session.KVWatch(watching, delivery.NodeLoginsBucket, node) + if werr != nil { + stop() + if line := "this node's login is not followed yet (" + werr.Error() + "): it dials as it last did, and looks again in " + loginsFollowAgain.String(); line != said { + said = line + say(line) + } + } else { + said = "" + for u := range updates { + if u.Caught || u.Deleted || len(u.Value) == 0 { + continue + } + nl, err := delivery.DecodeNodeLogin(u.Value) + if err != nil { + say("the key " + node + " of " + delivery.NodeLoginsRead + " is not read: " + err.Error()) + continue + } + hear(nl) + } + stop() + } + select { + case <-linked.Done(): + case <-time.After(loginsFollowAgain): + } + cancel() + } +} + +// loginSecretModule is the module the vault names this engine's login values under: own//mesh-host/ (ADR 0583 §3, +// insight). +const loginSecretModule = "mesh-host" + +// fetchLogin asks the vault, through its existing hand-out of a machine's own secrets, for the value of login, sealed to this +// node's key, and opens it with the sealing key; no new verb is asked for it. +func fetchLogin(queue *link.Session, statePath string) func(context.Context, string) (string, error) { + unseal := sealOpener(statePath) + return func(ctx context.Context, login string) (string, error) { + raw, err := queue.AskVault(ctx, "own", map[string]any{}) + if err != nil { + return "", err + } + var a struct { + Own []struct { + Module string `json:"module"` + Name string `json:"name"` + Sealed json.RawMessage `json:"sealed"` + } `json:"own"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return "", fmt.Errorf("the vault's own answer is not one: %w", err) + } + for _, o := range a.Own { + if o.Module != loginSecretModule || o.Name != login { + continue + } + sealed := string(o.Sealed) + var text string + if json.Unmarshal(o.Sealed, &text) == nil { + sealed = text + } + value, err := unseal(sealed) + if err != nil { + return "", fmt.Errorf("the value of %s cannot be opened with this node's sealing key: %w", login, err) + } + return string(value), nil + } + return "", fmt.Errorf("the vault hands this node no value for %s (own/%s/%s/%s)", login, queueNode(queue), loginSecretModule, login) + } +} + +func queueNode(q *link.Session) string { return q.Membership.Node } + +// newRotator is the engine's side of a rotation over this session (ADR 0583 §3-§6): the value fetched through the vault's own +// hand-out, the test-dial on a second connection, the confirmation over the new link, and a fall back carried in the health +// report. +func newRotator(queue *link.Session, statePath string, timeout time.Duration, say link.Announce) *link.Rotator { + r := &link.Rotator{Logins: queue.Membership.Logins, Say: say, + Fetch: fetchLogin(queue, statePath), + TestDial: func(ctx context.Context, login, password string) error { + return link.TestDial(ctx, queue.Membership, login, password, timeout) + }, + Confirm: func(ctx context.Context) error { + login, _ := queue.Membership.Logins.Current() + if err := queue.WhenLinkedAs(ctx, login); err != nil { + return err + } + _, err := queue.AskVault(ctx, "node-account-confirm", map[string]any{}) + return err + }} + r.Watch() + // The fall back is carried in this node's health report, from which the watchman raises node-account-fallback (ADR 0583 §6). + queue.Fallback = r.Fallback + return r +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 97d0bc6..fd601a3 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -30,6 +30,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/accounts" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" @@ -821,6 +822,11 @@ func enrol(ctx context.Context, opts options) error { // the bus's name, so every enrolment ended at "this token is for the \"\" bus" // (novox/hq 04-ISSUES/146). Nothing caught it because nothing had enrolled since the bus // changed. + // Through the seat mesh-enrolment when the bus serves it to this token (novox/hq ADR 0587 §6): the controller's enrolment + // subject below stays for as long as the controller holds its lease. + if handled, err := enrolThroughSeat(ctx, opts, token, *name, mine, sealing, serving, reported, found, identityPath); handled || err != nil { + return err + } reply, err := link.Enrol(ctx, link.Approach{Address: token.Broker, Fingerprint: token.Fingerprint, Transport: link.OnNATS}, *name, token.Secret, @@ -1074,6 +1080,8 @@ func runLink(ctx context.Context, opts options) error { Password: mine.Membership.Password, Transport: mine.Membership.Transport, Signer: mine.Membership.Signer, + // Which of its two logins it dials as (ADR 0583 §1): the one the vault's node-logins names. + Logins: loadLogins(opts.state, mine.Node, mine.Membership.Password), } // What follows an apply of a module (the node agent's, novox/hq ADR 0539): a membership it delivered for another bus @@ -1190,6 +1198,14 @@ func runLink(ctx context.Context, opts options) error { // And what it trusts, followed on the bus, said in its record whenever it changes (ADR 0550 §2). trust.onChange(func() { factsOfNow(aside, current.get()) }) go followTrust(aside, queue, trust, say) + // And the login it dials as, which the vault's node-logins names for this node (novox/hq ADR 0583 §1). + rotator := newRotator(queue, opts.state, opts.timeout, say) + go followLogins(aside, queue, mine.Node, func(nl delivery.NodeLogin) { + if line := membership.Logins.Heard(nl); line != "" { + say(line) + } + rotator.Heard(aside, nl) + }, say) go nodeAgent(aside, opts, mine.Membership.Signer, trust, queue, sched, say, afterApply, current, factsOfNow) go busyPasses(aside, func() { if ag := current.get(); ag != nil { diff --git a/go.mod b/go.mod index d657e1e..5ed422a 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102004146-dff0ce2abb21 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 diff --git a/go.sum b/go.sum index c4ff816..5937b7a 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4 h1:xeuMBwOH git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101151945-b2a27ed5cec4/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c h1:/YCxh+NYfqry3PxZYY1nQXaccupobxpsFt8c9BYuGBg= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102004146-dff0ce2abb21 h1:4I4ayo0fZYP4PBfEX1sGDDd6arnw5IQJySJ3jcFobPM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102004146-dff0ce2abb21/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= diff --git a/internal/identity/identity_test.go b/internal/identity/identity_test.go index 9dd7293..8a1e4b6 100644 --- a/internal/identity/identity_test.go +++ b/internal/identity/identity_test.go @@ -235,7 +235,9 @@ func TestTheWireFormatIsExactlyTheseFieldNames(t *testing.T) { } } -func TestACompleteTokenParses(t *testing.T) { +// A complete token parses, a pasted one tolerates whitespace, and one of another version is refused (three tests merged; the +// token's shape is mesh-sdk's since ADR 0587 §3). +func TestACompleteTokenParsesAPastedOneToleratesWhitespaceAndAnotherVersionIsRefused(t *testing.T) { got, err := ParseToken(completeToken(t)) if err != nil { t.Fatal(err) @@ -243,12 +245,12 @@ func TestACompleteTokenParses(t *testing.T) { if got.Broker != "192.0.2.10:5671" || len(got.SignerKey()) != ed25519.PublicKeySize { t.Errorf("parsed %+v", got) } -} - -func TestAPastedTokenTolerantOfWhitespace(t *testing.T) { if _, err := ParseToken(" " + completeToken(t) + "\n"); err != nil { t.Errorf("a pasted token was refused: %v", err) } + if _, err := ParseToken(encodeToken(t, `{"v":99,"broker":"b","fingerprint":"f","secret":"s"}`)); err == nil { + t.Fatal("a token from an unknown version was accepted") + } } func TestAnIncompleteTokenIsRefusedWholeAndSaysWhatIsMissing(t *testing.T) { @@ -272,12 +274,6 @@ func TestAnIncompleteTokenIsRefusedWholeAndSaysWhatIsMissing(t *testing.T) { } } -func TestATokenFromAnotherVersionIsRefused(t *testing.T) { - if _, err := ParseToken(encodeToken(t, `{"v":99,"broker":"b","fingerprint":"f","secret":"s"}`)); err == nil { - t.Fatal("a token from an unknown version was accepted") - } -} - func TestGarbageIsRefused(t *testing.T) { for _, bad := range []string{"", "!!!not base64!!!", "aGVsbG8"} { if _, err := ParseToken(bad); err == nil { @@ -350,7 +346,8 @@ func TestASealingKeyOpensOnlyWhatWasSealedToIt(t *testing.T) { } } -func TestSealingTheSameValueTwiceLooksDifferent(t *testing.T) { +// Sealing is randomised and a node with no sealing key is told what to do (two tests merged). +func TestSealingIsRandomisedAndANodeWithNoSealingKeySaysWhatToDo(t *testing.T) { // Sealed boxes are randomised, so an observer cannot tell that two nodes were given the same // password, nor that a rotation changed nothing. Worth asserting because the alternative is // a subtle leak nobody would look for. @@ -360,9 +357,6 @@ func TestSealingTheSameValueTwiceLooksDifferent(t *testing.T) { if first == second { t.Fatal("sealing is deterministic, so equal secrets are visible as equal blobs") } -} - -func TestANodeWithNoSealingKeySaysWhatToDo(t *testing.T) { // Rather than making one. A key the mesh was never told about is a key nothing can be sealed // to, so a node that quietly created one would look fine and receive nothing for ever. _, err := LoadSealingKey(t.TempDir() + "/absent.key") diff --git a/internal/identity/token.go b/internal/identity/token.go index 5241334..f024bb8 100644 --- a/internal/identity/token.go +++ b/internal/identity/token.go @@ -1,112 +1,15 @@ package identity -import ( - "crypto/ed25519" - "encoding/base64" - "encoding/json" - "fmt" - "strings" -) +import "git.novox.be/novox/mesh-sdk/go/delivery" -// Token is what a person carries to a machine that is joining. -// -// novox/hq ADR 0004 — four things: where the broker is, what certificate to expect there, whose -// signature to believe afterwards, and a one-time right to join. -// -// THIS IS A WIRE FORMAT SHARED WITH THE CONTROL PLANE, which writes it. The two definitions are -// separate on purpose — the host requires nothing present and does not import the control plane — -// so they are held together by a test on each side asserting the exact field names rather than by -// a shared type. If a field is renamed here and not there, that test fails on both sides. -type Token struct { - Version int `json:"v"` +// Token is what a person carries to a machine that is joining (novox/hq ADR 0004): where the broker is, what certificate to +// expect there, whose signature to believe afterwards, and a one-time right to join. Its shape and its reading are mesh-sdk's +// (delivery.Token), which the vault that makes it, mesh-cli that opens it and this engine that parses it share (novox/hq ADR +// 0587 §3, issue 774); these are the names this package has always had for them. +type Token = delivery.Token - // Node is what the mesh calls this machine, and it arrives here because the node cannot work - // it out. The broker account it must authenticate as is named after it, so it has to be known - // before the mesh can say anything — and without it enrolment is a connection refused with an - // empty username, which names nothing. - Node string `json:"node,omitempty"` +// TokenTunnel is the joining machine's side of its first tunnel (delivery.TokenTunnel). +type TokenTunnel = delivery.TokenTunnel - Broker string `json:"broker,omitempty"` - Fingerprint string `json:"fingerprint,omitempty"` - Signer []byte `json:"signer,omitempty"` - Secret string `json:"secret"` - - // Adopted says this node joins adopted (novox/hq ADR 0100). The host checks it speaks the - // firewall found here before enrolling, because an adopted node keeps that firewall in force. - // Absent for a converged node. - Adopted bool `json:"adopted,omitempty"` - - // Tunnel is this machine's first tunnel, when the token was issued for the key it made with - // `key` (novox/hq ADR 0169): its own address and the hub to reach. It brings the tunnel up from - // this alone and reaches the bus over it, so the bus never has to face the internet. - Tunnel *TokenTunnel `json:"tunnel,omitempty"` -} - -// TokenTunnel is the joining machine's side of its first tunnel. Field names are the wire format -// the controller writes. -type TokenTunnel struct { - Key string `json:"key"` - Address string `json:"address"` - Range string `json:"range"` - HubKey string `json:"hub_key"` - HubEndpoint string `json:"hub_endpoint"` -} - -// ParseToken reads a token a person pasted. -// -// Every refusal here says *this is not a token* rather than *this is the wrong token*. The -// difference matters once there is a mesh: a host must tell "this is not from the mesh I joined" -// apart from "this is malformed" (novox/hq ADR 0004), and the first is a signature check later, -// not a parse failure here. -func ParseToken(encoded string) (Token, error) { - raw, err := base64.RawURLEncoding.DecodeString(strings.TrimSpace(encoded)) - if err != nil { - return Token{}, fmt.Errorf("this is not a token: %w", err) - } - var t Token - if err := json.Unmarshal(raw, &t); err != nil { - return Token{}, fmt.Errorf("this is not a token: %w", err) - } - if t.Version != 1 { - return Token{}, fmt.Errorf( - "this token says it is version %d, and this host understands version 1", t.Version) - } - - var missing []string - if strings.TrimSpace(t.Broker) == "" { - missing = append(missing, "the broker's address") - } - if strings.TrimSpace(t.Fingerprint) == "" { - missing = append(missing, "the broker certificate's fingerprint") - } - if len(t.Signer) != ed25519.PublicKeySize { - missing = append(missing, "the control plane's signing key") - } - if strings.TrimSpace(t.Secret) == "" { - missing = append(missing, "the one-time secret") - } - if tt := t.Tunnel; tt != nil { - for _, part := range []struct{ value, says string }{ - {tt.Key, "the tunnel key it was issued for"}, {tt.Address, "this machine's address"}, - {tt.Range, "the private network's range"}, {tt.HubKey, "the hub's tunnel key"}, - {tt.HubEndpoint, "where the hub's tunnel is dialled"}, - } { - if strings.TrimSpace(part.value) == "" { - missing = append(missing, part.says) - } - } - } - if len(missing) > 0 { - // Refused whole rather than used partially. A token missing the fingerprint would have - // this node connect to whatever answers at that address, and one missing the signing key - // would leave it unable to tell a declaration from a forgery — so an incomplete token is - // not a reduced capability, it is an unsafe one. - return Token{}, fmt.Errorf( - "this token is missing %s, so it cannot be used to join anything", - strings.Join(missing, ", ")) - } - return t, nil -} - -// SignerKey is the control plane's public signing key, as a key. -func (t Token) SignerKey() ed25519.PublicKey { return ed25519.PublicKey(t.Signer) } +// ParseToken reads a token a person pasted (delivery.ParseToken). +func ParseToken(encoded string) (Token, error) { return delivery.ParseToken(encoded) } diff --git a/internal/link/enrol_seat.go b/internal/link/enrol_seat.go new file mode 100644 index 0000000..c5d7eb7 --- /dev/null +++ b/internal/link/enrol_seat.go @@ -0,0 +1,241 @@ +package link + +import ( + "context" + "crypto/ed25519" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Joining a mesh through the seat mesh-enrolment (novox/hq ADR 0587 §6, issue 774): the node-engine dials the bus as enrol. +// with the certificate's fingerprint pinned, calls `redeem` with the public halves of the keys it made where it runs and a +// proof signed with its identity key, refuses an answer not signed by the key its token carries, keeps its membership, dials +// again as the login node-logins names for it, and calls `confirm`. The controller's enrolment subject (Enrol) stays for as long +// as the controller holds its lease. + +// EnrolmentSeat is the seat that adopts a new machine, and its two verbs a node-engine calls. +const ( + EnrolmentSeat = "mesh-enrolment" + EnrolmentRedeem = "redeem" + EnrolmentConfirm = "confirm" +) + +// CalledSeatSubject is where a node-engine calls a seat's verb in its own name: the caller's user, each dot written `~`. +func CalledSeatSubject(seat, verb, user string) string { + return "mesh.seat." + seat + ".call." + verb + "." + strings.ReplaceAll(user, ".", "~") +} + +// ErrNoSeat is the bus saying nothing serves the seat for this caller: the enrolment user may not publish the seat's call (the +// controller still holds its lease) or no holder answers it. +var ErrNoSeat = errors.New("the seat mesh-enrolment is not served to this machine's token") + +// VerifyRedeemAnswer holds an answer to what it must be before anything is kept (ADR 0587 §6): it names the node asked for, +// carries the signing key the token carries, and is signed by that key under the context of an enrolment's answer. An answer +// signed by another key is refused, and so is one that names another signer. +func VerifyRedeemAnswer(a delivery.RedeemAnswer, node string, signer ed25519.PublicKey) error { + if a.Node != node { + return fmt.Errorf("the answer is for %q, and this machine asked as %q", a.Node, node) + } + if a.Signer != base64.StdEncoding.EncodeToString(signer) { + return errors.New("the answer names a signing key other than the one the token carries") + } + sig, err := base64.StdEncoding.DecodeString(a.Signature) + if err != nil || len(sig) == 0 { + return errors.New("the answer is not signed") + } + kindContext, err := delivery.SignContext(delivery.KindEnrolment) + if err != nil { + return err + } + if _, err := delivery.VerifyBy([]delivery.Signer{{Name: "the token's", Key: signer}}, kindContext, a.SignedBytes(), sig); err != nil { + return fmt.Errorf("the answer is not signed by the key the token carries: %w", err) + } + if a.Login == "" || a.Account == "" { + return errors.New("the answer names no login or no account") + } + return nil +} + +// seatAnswer reads a seat call's reply: its result, or its refusal as an error. +func seatAnswer(reply []byte) (json.RawMessage, error) { + var env struct { + Result json.RawMessage `json:"result"` + Error string `json:"error"` + } + if err := json.Unmarshal(reply, &env); err != nil { + return nil, errors.New("the seat answered with something that is not a tool's answer") + } + if env.Error != "" { + return nil, errors.New(env.Error) + } + return env.Result, nil +} + +// Redeem calls the seat's redeem as enrol., with the token's secret as the bus password, on the pinned certificate. It +// asks again with the same keys while the holder says to (a stopped redeem finishes; the bus node loads the account), for +// as long as a machine asking the controller would (EnrolPatience). An answer is returned as the holder gave it: the caller +// verifies it (VerifyRedeemAnswer) before it keeps anything. +func Redeem(ctx context.Context, to Approach, node, secret string, req delivery.Redeem, timeout time.Duration) (delivery.RedeemAnswer, error) { + if to.Transport != OnNATS { + return delivery.RedeemAnswer{}, fmt.Errorf("this token is for the %q bus, and the mesh's bus is %s", to.Transport, OnNATS) + } + config, err := PinnedConfig(to.Fingerprint) + if err != nil { + return delivery.RedeemAnswer{}, err + } + user := delivery.EnrolmentUser(node) + conn, err := nats.Connect(natsURL(to.Address), nats.Secure(config), nats.CustomInboxPrefix("_INBOX."+user), + nats.UserInfo(user, secret), nats.Name("mesh-host/enrol/"+node), nats.Timeout(timeout), nats.NoReconnect()) + if err != nil { + if errors.Is(err, ErrWrongCertificate) { + return delivery.RedeemAnswer{}, err + } + return delivery.RedeemAnswer{}, fmt.Errorf("cannot reach the bus at %s as %s: %w", to.Address, user, err) + } + defer conn.Close() + body, err := json.Marshal(req) + if err != nil { + return delivery.RedeemAnswer{}, err + } + began := time.Now() + for { + msg, err := conn.RequestWithContext(ctx, CalledSeatSubject(EnrolmentSeat, EnrolmentRedeem, user), body) + switch { + case errors.Is(err, nats.ErrNoResponders) || (err != nil && strings.Contains(err.Error(), "permissions violation")): + return delivery.RedeemAnswer{}, fmt.Errorf("%w: %v", ErrNoSeat, err) + case err != nil: + return delivery.RedeemAnswer{}, fmt.Errorf("the seat did not answer redeem: %w", err) + } + result, aerr := seatAnswer(msg.Data) + if aerr != nil { + if strings.Contains(aerr.Error(), "call redeem again with the same keys") && time.Since(began) < EnrolPatience { + select { + case <-ctx.Done(): + return delivery.RedeemAnswer{}, ctx.Err() + case <-time.After(AskAgainAfter): + continue + } + } + return delivery.RedeemAnswer{}, fmt.Errorf("%w: %v", ErrRefused, aerr) + } + var answer delivery.RedeemAnswer + if err := json.Unmarshal(result, &answer); err != nil { + return delivery.RedeemAnswer{}, fmt.Errorf("the seat's answer to redeem is not one: %w", err) + } + return answer, nil + } +} + +// Confirming is this machine dialled as its own login after it redeemed: it reads its key of node-logins and calls confirm. +type Confirming interface { + // Login is this node's key of the vault's node-logins. + Login(ctx context.Context) (delivery.NodeLogin, error) + // Confirm calls the seat's confirm; the runtime's proof of the caller as node. is the confirmation (ADR 0583 §3). + Confirm(ctx context.Context) error + Close() +} + +// Joining is the engine's side of a redeem, in the order ADR 0587 §6 gives it, over what reaches the bus and the disk. +type Joining struct { + Node string + Signer ed25519.PublicKey + // Redeem calls the seat; Open opens the account sealed to this node's sealing key; Keep writes the membership, the login + // and the keys down; Dial dials the bus as login and answers it ready to read node-logins and confirm. + Redeem func(ctx context.Context) (delivery.RedeemAnswer, error) + Open func(a delivery.RedeemAnswer) (string, error) + Keep func(a delivery.RedeemAnswer, password string) error + Dial func(ctx context.Context, login, password string) (Confirming, error) +} + +// Run joins: the answer is verified before anything is kept (an answer not signed by the key the token carries keeps nothing), +// kept, and then the machine dials as the login its key of node-logins names and confirms. Nothing is applied here: the engine +// applies once its placement names a module. +func (j Joining) Run(ctx context.Context) (delivery.RedeemAnswer, error) { + answer, err := j.Redeem(ctx) + if err != nil { + return delivery.RedeemAnswer{}, err + } + if err := VerifyRedeemAnswer(answer, j.Node, j.Signer); err != nil { + return delivery.RedeemAnswer{}, fmt.Errorf("%w: %v; nothing was kept", ErrRefused, err) + } + password, err := j.Open(answer) + if err != nil { + return delivery.RedeemAnswer{}, fmt.Errorf("the account in the answer cannot be opened: %w; nothing was kept", err) + } + if err := j.Keep(answer, password); err != nil { + return delivery.RedeemAnswer{}, fmt.Errorf("the mesh accepted this node as %q and what it was given could not be kept: %w\n"+ + "That token is spent, so getting back needs a new one", answer.Node, err) + } + c, err := j.Dial(ctx, answer.Login, password) + if err != nil { + return answer, fmt.Errorf("this node is known to the mesh as %q and cannot dial as %s yet: %w", answer.Node, answer.Login, err) + } + defer c.Close() + nl, err := c.Login(ctx) + if err != nil { + return answer, fmt.Errorf("this node's key of node-logins cannot be read: %w", err) + } + if nl.Login != answer.Login { + return answer, fmt.Errorf("node-logins names %s for this node and the answer gave it %s: it is not confirmed", nl.Login, answer.Login) + } + if err := c.Confirm(ctx); err != nil { + return answer, fmt.Errorf("the enrolment is redeemed and not confirmed (a person ends it with mesh-enrolment.revoke): %w", err) + } + return answer, nil +} + +// confirmingNats is a node dialled as its own login on the pinned certificate. +type confirmingNats struct { + conn *nats.Conn + node string + user string +} + +// DialConfirming dials the bus as login with password, on the pinned certificate, ready to read node-logins and confirm. +func DialConfirming(ctx context.Context, m Membership, login, password string, timeout time.Duration) (Confirming, error) { + config, err := PinnedConfig(m.Fingerprint) + if err != nil { + return nil, err + } + conn, err := nats.Connect(natsURL(m.Broker), nats.Secure(config), nats.UserInfo(login, password), + nats.CustomInboxPrefix("_INBOX.node."+m.Node), nats.Name("mesh-host/"+m.Node+"/enrol"), nats.Timeout(timeout), nats.NoReconnect()) + if err != nil { + return nil, fmt.Errorf("cannot dial as %s: %w", login, err) + } + return &confirmingNats{conn: conn, node: m.Node, user: login}, nil +} + +func (c *confirmingNats) Login(ctx context.Context) (delivery.NodeLogin, error) { + js, err := c.conn.JetStream() + if err != nil { + return delivery.NodeLogin{}, err + } + kv, err := js.KeyValue(delivery.NodeLoginsBucket) + if err != nil { + return delivery.NodeLogin{}, err + } + e, err := kv.Get(c.node) + if err != nil { + return delivery.NodeLogin{}, err + } + return delivery.DecodeNodeLogin(e.Value()) +} + +func (c *confirmingNats) Confirm(ctx context.Context) error { + msg, err := c.conn.RequestWithContext(ctx, CalledSeatSubject(EnrolmentSeat, EnrolmentConfirm, c.user), []byte(`{}`)) + if err != nil { + return err + } + _, err = seatAnswer(msg.Data) + return err +} + +func (c *confirmingNats) Close() { c.conn.Close() } diff --git a/internal/link/enrol_seat_test.go b/internal/link/enrol_seat_test.go new file mode 100644 index 0000000..b1393a9 --- /dev/null +++ b/internal/link/enrol_seat_test.go @@ -0,0 +1,104 @@ +package link + +import ( + "bytes" + "context" + "crypto/ed25519" + "encoding/base64" + "errors" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// signedBy is an answer for ace, signed under the enrolment context by the key seeded, naming that key as its signer. +func signedBy(t *testing.T, seed byte) (delivery.RedeemAnswer, ed25519.PublicKey) { + t.Helper() + priv := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{seed}, ed25519.SeedSize)) + pub := priv.Public().(ed25519.PublicKey) + a := delivery.RedeemAnswer{Node: "ace", Login: "node.ace", Account: `{"sealed":"box"}`, Certificate: "PEM", Signer: base64.StdEncoding.EncodeToString(pub)} + ctx, err := delivery.SignContext(delivery.KindEnrolment) + if err != nil { + t.Fatal(err) + } + a.Signature = base64.StdEncoding.EncodeToString(delivery.SignMessage(priv, ctx, a.SignedBytes())) + return a, pub +} + +type standInConfirming struct { + steps *[]string + nl delivery.NodeLogin +} + +func (c standInConfirming) Login(context.Context) (delivery.NodeLogin, error) { + *c.steps = append(*c.steps, "read node-logins") + return c.nl, nil +} +func (c standInConfirming) Confirm(context.Context) error { + *c.steps = append(*c.steps, "confirm") + return nil +} +func (c standInConfirming) Close() {} + +// 774/S64/AC1 and AC2 (ADR 0587 §6): an answer not signed by the key the token carries (or naming another) is refused and +// nothing is kept or dialled; a signed one is kept, then the engine dials as the login it names, reads its key of node-logins, +// and confirms; a key naming another login than the answer's is not confirmed. +func TestTheEngineKeepsOnlyAnAnswerSignedByTheKeyItsTokenCarries(t *testing.T) { + ctx := context.Background() + good, pub := signedBy(t, 1) + forged, _ := signedBy(t, 2) + badNode := good + badNode.Node = "g14" + for name, c := range map[string]struct { + answer delivery.RedeemAnswer + want string + }{ + "signed by another key": {forged, "signing key other than the one the token carries"}, + "another node": {badNode, `is for "g14"`}, + "altered after signing": {func() delivery.RedeemAnswer { a := good; a.Account = "other"; return a }(), "not signed by the key the token carries"}, + } { + var steps []string + j := Joining{Node: "ace", Signer: pub, + Redeem: func(context.Context) (delivery.RedeemAnswer, error) { return c.answer, nil }, + Open: func(delivery.RedeemAnswer) (string, error) { steps = append(steps, "open"); return "pw", nil }, + Keep: func(delivery.RedeemAnswer, string) error { steps = append(steps, "keep"); return nil }, + Dial: func(context.Context, string, string) (Confirming, error) { + steps = append(steps, "dial") + return nil, errors.New("must not dial") + }} + if _, err := j.Run(ctx); err == nil || !strings.Contains(err.Error(), c.want) || len(steps) != 0 { + t.Errorf("%s: %v, steps %v", name, err, steps) + } + } + var steps []string + login := "node.ace" + j := Joining{Node: "ace", Signer: pub, + Redeem: func(context.Context) (delivery.RedeemAnswer, error) { + steps = append(steps, "redeem") + return good, nil + }, + Open: func(delivery.RedeemAnswer) (string, error) { steps = append(steps, "open"); return "pw", nil }, + Keep: func(a delivery.RedeemAnswer, pw string) error { + steps = append(steps, "keep "+a.Login+" "+pw) + return nil + }, + Dial: func(_ context.Context, l, pw string) (Confirming, error) { + steps = append(steps, "dial "+l+" "+pw) + return standInConfirming{steps: &steps, nl: delivery.NodeLogin{Node: "ace", Login: login}}, nil + }} + if _, err := j.Run(ctx); err != nil { + t.Fatal(err) + } + if want := []string{"redeem", "open", "keep node.ace pw", "dial node.ace pw", "read node-logins", "confirm"}; !slices.Equal(steps, want) { + t.Errorf("steps %v, want %v", steps, want) + } + steps, login = nil, "node.ace--b" + if _, err := j.Run(ctx); err == nil || slices.Contains(steps, "confirm") { + t.Errorf("a key naming another login was confirmed: %v, %v", err, steps) + } + if CalledSeatSubject("mesh-enrolment", "redeem", "enrol.ace") != "mesh.seat.mesh-enrolment.call.redeem.enrol~ace" { + t.Error("the call's subject does not name its caller as the composer grants it") + } +} diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index afc81b4..06cedf2 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -59,6 +59,8 @@ type natsLink struct { engine micro.Service warrants *nats.Subscription node string + // user is the bus user this link authenticated as (node. or node.--b, ADR 0583 §1). + user string // refused carries each subject the bus refused this link a subscription to (RefusedSubscription), which serve drops // the link for after a pause, so every subscription is asked again on the next. refused chan string @@ -77,14 +79,16 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e if err != nil { return nil, err } + user, password := m.Credential() opts := []nats.Option{ nats.Secure(config), // The mesh names a machine's bus user "node." (the controller's principal scheme), and // the server refused the bare name the first time a machine dialled it: "authentication // error - User". The same string the mesh composed into the user list, or nothing connects. - nats.UserInfo("node."+m.Node, m.Password), + nats.UserInfo(user, password), // Replies to what this client asks the server arrive on its inbox, and the mesh grants a // machine exactly its own: the same prefix the user list was composed with. + // Whichever of its two logins it dials as (ADR 0583 §1). nats.CustomInboxPrefix("_INBOX.node." + m.Node), nats.Name("mesh-host/" + m.Node), nats.Timeout(timeout), @@ -99,6 +103,7 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e } conn, err := nats.Connect(natsURL(m.Broker), opts...) + m.Logins.dialed(user, err) if err != nil { if errors.Is(err, ErrWrongCertificate) { return nil, err @@ -112,7 +117,7 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e } l := &natsLink{ - conn: conn, js: js, node: m.Node, signer: m.Signer, + conn: conn, js: js, node: m.Node, user: user, signer: m.Signer, lost: make(chan error, 1), asked: make(chan struct{}, 1), refused: make(chan string, 64), @@ -381,3 +386,6 @@ func RefusedSubscription(err error) (string, bool) { } return m[1], true } + +// User is the bus user this link authenticated as. +func (l *natsLink) User() string { return l.user } diff --git a/internal/link/logins.go b/internal/link/logins.go new file mode 100644 index 0000000..1ee7fa7 --- /dev/null +++ b/internal/link/logins.go @@ -0,0 +1,210 @@ +package link + +import ( + "errors" + "strings" + "sync" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Logins is which of its two logins this node-engine dials the bus as, and the password of each it holds (novox/hq ADR 0583 §1, +// issue 774). A node's bus account alternates between node. and node.--b with the node's permissions and its inbox +// _INBOX.node. unchanged; the login in use is the vault's state node-logins' key of this node, which the engine reads and +// follows. Nothing here fixes the login: a membership with no Logins dials node. with its own password, as before any +// rotation. The passwords are the engine's own copies (the old one is kept until the vault says the rotation is done, ADR +// 0583 §4); Save is called whenever they change, so a restart dials as the login in use. +type Logins struct { + mu sync.Mutex + node string + current string + values map[string]string + save func(LoginsState) error + changed chan struct{} + // OnDial is told how each dial as a login went: authenticated, or refused for its credentials (the Rotator's fallback). + OnDial func(user string, authenticated bool) +} + +// LoginsState is what the engine keeps of its logins between runs. +type LoginsState struct { + // Current is the login in use ("" is node.) and Values the password of each login held. + Current string `json:"current,omitempty"` + Values map[string]string `json:"values,omitempty"` +} + +// NewLogins is the logins of node: first is the password of node. (the membership's), saved what was kept last time (none +// on a first run), and save writes the state down; nil keeps nothing between runs. +func NewLogins(node, first string, saved LoginsState, save func(LoginsState) error) *Logins { + l := &Logins{node: node, current: saved.Current, values: map[string]string{}, save: save, changed: make(chan struct{}, 1)} + for login, password := range saved.Values { + l.values[login] = password + } + if first != "" { + if _, ok := l.values[delivery.FirstLogin(node)]; !ok { + l.values[delivery.FirstLogin(node)] = first + } + } + if l.current == "" { + l.current = delivery.FirstLogin(node) + } + return l +} + +// Current is the login in use and its password. +func (l *Logins) Current() (login, password string) { + l.mu.Lock() + defer l.mu.Unlock() + return l.current, l.values[l.current] +} + +// Value is the password held for login. +func (l *Logins) Value(login string) (string, bool) { + l.mu.Lock() + defer l.mu.Unlock() + v, ok := l.values[login] + return v, ok && v != "" +} + +// Hold keeps the password of login, written down at once; a login of another node is refused. +func (l *Logins) Hold(login, password string) error { + l.mu.Lock() + defer l.mu.Unlock() + if n, ok := delivery.LoginNode(login); !ok || n != l.node { + return &loginError{login: login, node: l.node} + } + l.values[login] = password + return l.persist() +} + +// Use makes login the one dialled next, when its password is held; the next dial then uses it. +func (l *Logins) Use(login string) error { + l.mu.Lock() + defer l.mu.Unlock() + if v := l.values[login]; v == "" { + return &loginError{login: login, node: l.node, unheld: true} + } + if l.current != login { + l.current = login + select { + case l.changed <- struct{}{}: + default: + } + } + return l.persist() +} + +// Changed says the login in use changed since the last time it was read: the link open is dropped, and opened again as it. +func (l *Logins) Changed() <-chan struct{} { return l.changed } + +// Drop forgets the password of login (never the one in use), written down at once. +func (l *Logins) Drop(login string) error { + l.mu.Lock() + defer l.mu.Unlock() + if login == l.current { + return &loginError{login: login, node: l.node, inUse: true} + } + delete(l.values, login) + return l.persist() +} + +// Held is every login whose password is held. +func (l *Logins) Held() []string { + l.mu.Lock() + defer l.mu.Unlock() + var out []string + for login := range l.values { + out = append(out, login) + } + return out +} + +func (l *Logins) persist() error { + if l.save == nil { + return nil + } + state := LoginsState{Current: l.current, Values: map[string]string{}} + for k, v := range l.values { + state.Values[k] = v + } + return l.save(state) +} + +type loginError struct { + login, node string + unheld, inUse bool +} + +func (e *loginError) Error() string { + switch { + case e.unheld: + return "this engine holds no password for the login " + e.login + case e.inUse: + return "the login " + e.login + " is the one in use, and its password is not dropped" + } + return e.login + " is no login of the node " + e.node +} + +// Credential is the bus user and password a dial of m uses: the login in use when m names its logins, the node's own +// node. and the membership's password otherwise (the inbox, _INBOX.node., is the node's whichever login it is). +func (m Membership) Credential() (user, password string) { + if m.Logins != nil { + return m.Logins.Current() + } + return "node." + m.Node, m.Password +} + +// User is the bus user a dial of m uses. +func (m Membership) User() string { + user, _ := m.Credential() + return user +} + +// Heard takes this node's key of the vault's node-logins as read (a key read, or the empty value of a key gone): the login +// in use becomes the one it names when its password is held. A key naming a login whose password is not held changes +// nothing and says so; no key leaves the login as it is. +func (l *Logins) Heard(nl delivery.NodeLogin) (said string) { + if nl.Node != l.node || nl.Login == "" { + return "" + } + cur, _ := l.Current() + if nl.Login == cur { + return "" + } + if rotatingToUs(nl, cur) { + return "" // the rotation moves the engine, which switches after its test-dial (Rotator) + } + if err := l.Use(nl.Login); err != nil { + return "node-logins names " + nl.Login + " in use and this engine cannot dial as it: " + err.Error() + } + return "node-logins names " + nl.Login + " in use: dialled as it from now" +} + +// rotatingToUs says the key's rotating login is the one in use already: the engine switched ahead of the vault's own state. +func rotatingToUs(nl delivery.NodeLogin, current string) bool { + return nl.Rotating != "" && nl.Rotating == current +} + +// dialed tells OnDial how a dial as user went; err is the dial's, nil for a connection made. +func (l *Logins) dialed(user string, err error) { + if l == nil || l.OnDial == nil { + return + } + switch { + case err == nil: + l.OnDial(user, true) + case refusedForCredentials(err): + l.OnDial(user, false) + } +} + +// refusedForCredentials says a failed dial was the bus refusing the user or its password, not the network failing. +func refusedForCredentials(err error) bool { + if errors.Is(err, nats.ErrAuthorization) || errors.Is(err, nats.ErrAuthExpired) { + return true + } + text := strings.ToLower(err.Error()) + return strings.Contains(text, "authorization violation") || strings.Contains(text, "authentication error") || + strings.Contains(text, "user authentication") +} diff --git a/internal/link/logins_test.go b/internal/link/logins_test.go new file mode 100644 index 0000000..38ffd41 --- /dev/null +++ b/internal/link/logins_test.go @@ -0,0 +1,49 @@ +package link + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 774/S61/AC1 and AC2 (ADR 0583 §1): with no node-logins key, or one that names the first login, the engine dials as node. +// with the membership's password as before; with a key naming node.--b in use it dials as that login (whose value it +// holds), with the node's one inbox; a login whose value it does not hold is not dialled, and said. +func TestTheLoginDialledIsTheOneNodeLoginsNames(t *testing.T) { + var saved LoginsState + l := NewLogins("ace", "first-secret", LoginsState{}, func(s LoginsState) error { saved = s; return nil }) + m := Membership{Node: "ace", Password: "first-secret", Logins: l} + if u, p := m.Credential(); u != "node.ace" || p != "first-secret" { + t.Fatalf("no key: %s %s", u, p) + } + if u, p := (Membership{Node: "ace", Password: "plain"}).Credential(); u != "node.ace" || p != "plain" { + t.Fatalf("no logins at all: %s %s", u, p) + } + if said := l.Heard(delivery.NodeLogin{Node: "ace", Login: "node.ace"}); said != "" || m.User() != "node.ace" { + t.Errorf("the first login named: %q, %s", said, m.User()) + } + if said := l.Heard(delivery.NodeLogin{Node: "ace", Login: "node.ace--b"}); !strings.Contains(said, "cannot dial as it") || m.User() != "node.ace" { + t.Errorf("a login held no value for: %q, %s", said, m.User()) + } + if err := l.Hold("node.ace--b", "second-secret"); err != nil { + t.Fatal(err) + } + if err := l.Hold("node.g14", "x"); err == nil { + t.Error("another node's login was kept") + } + if said := l.Heard(delivery.NodeLogin{Node: "ace", Login: "node.ace--b"}); !strings.Contains(said, "dialled as it") { + t.Errorf("said %q", said) + } + if u, p := m.Credential(); u != "node.ace--b" || p != "second-secret" || saved.Current != "node.ace--b" { + t.Errorf("node-logins names --b: %s %s, kept %+v", u, p, saved) + } + // A restart dials as the login kept, and the engine's own first password stays held until the vault says done. + again := NewLogins("ace", "first-secret", saved, nil) + if u, _ := again.Current(); u != "node.ace--b" || len(again.Held()) != 2 { + t.Errorf("after a restart: %s, held %v", u, again.Held()) + } + if CalledVaultSubjectAs("own", m.User()) != "mesh.seat.mesh-vault.call.own.node~ace--b" { + t.Errorf("the vault is asked in the name of the login in use: %s", CalledVaultSubjectAs("own", m.User())) + } +} diff --git a/internal/link/movesecret.go b/internal/link/movesecret.go index 4342796..cb17df0 100644 --- a/internal/link/movesecret.go +++ b/internal/link/movesecret.go @@ -175,7 +175,13 @@ func NodeEngineUser(node string) string { return "node." + node } // CalledVaultSubject is where this node's engine asks one of the vault's verbs in its own name: each dot of its // user's name written `~`, as the controller's broker writes a caller's token. func CalledVaultSubject(verb, node string) string { - return "mesh.seat." + VaultSeat + ".call." + verb + "." + strings.ReplaceAll(NodeEngineUser(node), ".", "~") + return CalledVaultSubjectAs(verb, NodeEngineUser(node)) +} + +// CalledVaultSubjectAs is CalledVaultSubject for the login the engine is dialled as (node. or node.--b, ADR 0583 §1): +// the bus lets only that user publish a subject naming it. +func CalledVaultSubjectAs(verb, user string) string { + return "mesh.seat." + VaultSeat + ".call." + verb + "." + strings.ReplaceAll(user, ".", "~") } // VaultWithin is how long the engine gives the vault to answer one step. @@ -201,7 +207,7 @@ func (q *Session) AskVault(ctx context.Context, verb string, args any) (json.Raw } asking, cancel := context.WithTimeout(ctx, VaultWithin) defer cancel() - reply, err := asker.Ask(asking, CalledVaultSubject(verb, q.Membership.Node), body) + reply, err := asker.Ask(asking, CalledVaultSubjectAs(verb, q.Membership.User()), body) if err != nil { return nil, fmt.Errorf("the vault's %s could not be asked: %w", verb, err) } diff --git a/internal/link/nodehealth.go b/internal/link/nodehealth.go index b938dbb..5aa2d87 100644 --- a/internal/link/nodehealth.go +++ b/internal/link/nodehealth.go @@ -26,7 +26,8 @@ import ( // read across every node's report; // - machine-units: Units, the failed system units no module owns; // - relogin-needed: Relogin, each account whose running session lacks a group it was given; -// - directory-used-as-found: Found, each directory a module declares that the engine uses as found. +// - directory-used-as-found: Found, each directory a module declares that the engine uses as found; +// - node-account-fallback: Fallback, the engine's fall back to its old login during a rotation (ADR 0583 §6). // // The report's layout is mesh-sdk's (delivery.NodeHealth): the bucket, its version and its parts. @@ -176,6 +177,9 @@ func (q *Session) WriteNodeHealth(ctx context.Context, h NodeHealth) (bool, erro if err != nil { return false, err } + if q.Fallback != nil && h.Fallback == nil { + h.Fallback = q.Fallback() // node-account-fallback (ADR 0583 §6) + } return writeNodeHealth(ctx, store, &q.healthWritten, h, time.Now) } diff --git a/internal/link/pinned_test.go b/internal/link/pinned_test.go index 05d6ec8..d3dbd6f 100644 --- a/internal/link/pinned_test.go +++ b/internal/link/pinned_test.go @@ -61,13 +61,21 @@ func server(t *testing.T) (address string, fingerprint string) { return listener.Addr().String(), Fingerprint(der) } -func TestTheRightBrokerIsAccepted(t *testing.T) { +// The right broker is accepted; and a malformed pin is caught before connecting, rather than at the handshake, so a mistyped +// token fails while a person is looking at it (two tests merged). +func TestTheRightBrokerIsAcceptedAndAMalformedPinRefused(t *testing.T) { address, pin := server(t) conn, err := dialPinned(address, pin, 5*time.Second) if err != nil { t.Fatalf("the broker its token describes was refused: %v", err) } conn.Close() + for _, bad := range []string{"", "sha256:short", strings.Repeat("a", 64), + "sha256:" + strings.Repeat("z", 64), "md5:" + strings.Repeat("a", 64)} { + if _, err := PinnedConfig(bad); err == nil { + t.Errorf("%q was accepted as a fingerprint", bad) + } + } } func TestADifferentBrokerIsRefused(t *testing.T) { @@ -138,17 +146,6 @@ func TestNothingIsSentToTheWrongBroker(t *testing.T) { } } -func TestAMalformedPinIsRefusedBeforeConnecting(t *testing.T) { - // Caught here rather than at the handshake, so a mistyped token fails while a person is - // looking at it. - for _, bad := range []string{"", "sha256:short", strings.Repeat("a", 64), - "sha256:" + strings.Repeat("z", 64), "md5:" + strings.Repeat("a", 64)} { - if _, err := PinnedConfig(bad); err == nil { - t.Errorf("%q was accepted as a fingerprint", bad) - } - } -} - func TestAnUnreachableBrokerIsAnOrdinaryFailure(t *testing.T) { // Must not read as a wrong certificate: one is a network problem worth retrying, the other // means the mesh was substituted. diff --git a/internal/link/rotate.go b/internal/link/rotate.go new file mode 100644 index 0000000..12d06bb --- /dev/null +++ b/internal/link/rotate.go @@ -0,0 +1,202 @@ +package link + +import ( + "context" + "fmt" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Rotator is the engine's side of a node's account rotation (novox/hq ADR 0583 §3-§6, issue 774). The vault begins a rotation +// and writes the node's key of node-logins: the login in use, the login rotating to and the stage. On that key's ready the +// engine fetches the new value, dials a second connection as the new login, and switches only if it authenticates; over the new +// connection it confirms. It keeps the old value until the vault says the rotation is done, switches back before the new value +// is dropped when the rotation is abandoned, and falls back to the old value after repeated authentication failures with the +// new one, saying so. Nothing here polls: every step answers a change of the key, or a failed dial. +type Rotator struct { + Logins *Logins + // Fetch opens the value of a login the vault hands this node as own//mesh-host/ (ADR 0583 §3, insight). + Fetch func(ctx context.Context, login string) (string, error) + // TestDial dials a second connection as login with password and answers whether it authenticated, closing it. + TestDial func(ctx context.Context, login, password string) error + // Confirm calls mesh-vault.node-account-confirm over the link as the login now in use, once it is open. + Confirm func(ctx context.Context) error + // Say is the engine's journal. + Say func(string) + // Now is the clock a fall back is dated by; time.Now when nil. + Now func() time.Time + // FailBound is how many authentication failures with the new login, before rotation-done, it falls back after; 3 when 0. + FailBound int + + mu sync.Mutex + old string // the login switched away from, while the rotation is under way + failures int + // fellBack is the key's `since` a fallback was made for: the same rotation is not tried again. + fellBackSince string + // fallen is the fall back the node's health report carries (delivery.LoginFallback), from which the watchman raises + // node-account-fallback (ADR 0583 §6), until the vault's key of this node says anything new. + fallen *delivery.LoginFallback + switchedSince string +} + +func (r *Rotator) say(f string, a ...any) { + if r.Say != nil { + r.Say(fmt.Sprintf(f, a...)) + } +} + +// Heard takes this node's key of node-logins as read. +func (r *Rotator) Heard(ctx context.Context, nl delivery.NodeLogin) { + r.mu.Lock() + defer r.mu.Unlock() + if r.fallen != nil && nl.Since.String() != r.fellBackSince { + r.fallen = nil // the vault moved on (abandoned, done, or another rotation begun): the fall back is no longer said + } + cur, _ := r.Logins.Current() + switch { + case nl.Rotating == "" && nl.Login == cur: + // Done (the switch happened and the old login is retired), or no rotation at all: the other login's value goes (§4), + // only now that the vault says so. + r.old, r.failures = "", 0 + if other := delivery.OtherLogin(nl.Node, cur); other != "" { + if _, held := r.Logins.Value(other); held { + if err := r.Logins.Drop(other); err != nil { + r.say("the old login %s's value is not dropped: %v", other, err) + } else { + r.say("rotation-done: the value of %s is dropped, %s is the login in use", other, cur) + } + } + } + case nl.Rotating == "": + // Abandoned after the switch (before or after the confirmation), or the vault moved on: back to the login the vault keeps, + // before the value of the one it drops. + if _, held := r.Logins.Value(nl.Login); !held { + r.say("the vault keeps %s in use and this engine holds no value for it: it stays on %s", nl.Login, cur) + return + } + if err := r.Logins.Use(nl.Login); err != nil { + r.say("cannot switch back to %s: %v", nl.Login, err) + return + } + r.say("the rotation is abandoned: this engine dials as %s again", nl.Login) + r.old, r.failures = "", 0 + if err := r.Logins.Drop(cur); err != nil { + r.say("the abandoned login %s's value is not dropped: %v", cur, err) + } + case nl.Login == cur && nl.Stage != delivery.RotationBegun: + if r.fellBackSince != "" && r.fellBackSince == nl.Since.String() { + return // fallen back for this rotation: not tried again until the vault begins another + } + r.switchTo(ctx, nl) + case nl.Rotating == cur && nl.Stage != delivery.RotationConfirmed: + r.confirm(ctx, nl.Rotating) + } +} + +// switchTo fetches the new value, test-dials it and, only if it authenticates, switches and confirms (§3). +func (r *Rotator) switchTo(ctx context.Context, nl delivery.NodeLogin) { + next := nl.Rotating + password, held := r.Logins.Value(next) + if !held { + var err error + if password, err = r.Fetch(ctx, next); err != nil { + r.say("the value of %s is not fetched, so this node stays on %s: %v", next, nl.Login, err) + return + } + if err := r.Logins.Hold(next, password); err != nil { + r.say("the value of %s is not kept, so this node stays on %s: %v", next, nl.Login, err) + return + } + } + if err := r.TestDial(ctx, next, password); err != nil { + r.say("the test-dial as %s did not authenticate, so this node stays on %s: %v", next, nl.Login, err) + return + } + if err := r.Logins.Use(next); err != nil { + r.say("this node cannot switch to %s: %v", next, err) + return + } + r.old, r.failures, r.switchedSince = nl.Login, 0, nl.Since.String() + r.say("the test-dial as %s authenticated: this node dials as %s from now, and keeps the value of %s until the vault says the rotation is done", + next, next, nl.Login) + r.confirm(ctx, next) +} + +func (r *Rotator) confirm(ctx context.Context, login string) { + if r.Confirm == nil { + return + } + if err := r.Confirm(ctx); err != nil { + r.say("node-account-confirm as %s is not answered yet (asked again when the key changes or the link opens): %v", login, err) + return + } + r.say("node-account-confirm as %s is answered", login) +} + +// AuthFailed is told of each failed authentication as login: before rotation-done, repeated failures with the new login (the +// one switched to) bring the engine back to the old value, saying node-account-fallback (§6). +func (r *Rotator) AuthFailed(login string) { + r.mu.Lock() + defer r.mu.Unlock() + cur, _ := r.Logins.Current() + if r.old == "" || login != cur { + return + } + r.failures++ + bound := r.FailBound + if bound == 0 { + bound = 3 + } + if r.failures < bound { + return + } + if err := r.Logins.Use(r.old); err != nil { + r.say("cannot fall back to %s: %v", r.old, err) + return + } + text := fmt.Sprintf("%s failed to authenticate %d times before the vault said rotation-done: this node dials as %s again", login, r.failures, r.old) + r.say("%s: %s", delivery.NodeAccountFallbackFinding, text) + now := time.Now + if r.Now != nil { + now = r.Now + } + r.fallen = &delivery.LoginFallback{Node: r.Logins.node, From: login, Why: text, At: now().UTC()} + r.old, r.failures = "", 0 + r.fellBackSince = r.switchedSince +} + +// Fallback is the fall back the node's health report carries now, nil when there is none. +func (r *Rotator) Fallback() *delivery.LoginFallback { + if r == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + if r.fallen == nil { + return nil + } + f := *r.fallen + return &f +} + +// AuthOK is told of each successful connection as login: the failures with it are forgotten. +func (r *Rotator) AuthOK(login string) { + r.mu.Lock() + defer r.mu.Unlock() + if cur, _ := r.Logins.Current(); login == cur { + r.failures = 0 + } +} + +// Watch makes the engine's dials tell r how they went: every refusal of the login in use counts towards the fallback. +func (r *Rotator) Watch() { + r.Logins.OnDial = func(user string, authenticated bool) { + if authenticated { + r.AuthOK(user) + } else { + r.AuthFailed(user) + } + } +} diff --git a/internal/link/rotate_integration_test.go b/internal/link/rotate_integration_test.go new file mode 100644 index 0000000..a3d42ea --- /dev/null +++ b/internal/link/rotate_integration_test.go @@ -0,0 +1,170 @@ +//go:build integration + +package link + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 774/S65/AC1 (ADR 0583 §2-§6): the engine's rotation against a real bus whose list of users a stand-in vault reloads step by +// step. At every step at least one login authenticates; the engine switches only once the list holds both logins; the +// confirmation arrives over the new connection, in the new login's name; at rotation-done the old login is retired and its +// value dropped; with the list never loaded the engine never switches; and an abandon after the switch restores the old login. +func TestARotationAgainstARealBusWhoseListIsReloadedStepByStep(t *testing.T) { + const first, second = "node.ace", "node.ace--b" + users := func(logins ...string) []*natsserver.User { + out := []*natsserver.User{{Username: "vault", Password: "vault-pw"}} + for _, l := range logins { + pw := map[string]string{first: "pw-first", second: "pw-second"}[l] + out = append(out, &natsserver.User{Username: l, Password: pw, Permissions: &natsserver.Permissions{ + Publish: &natsserver.SubjectPermission{Allow: []string{"mesh.seat.mesh-vault.call.>", "_INBOX.node.ace.>"}}, + Subscribe: &natsserver.SubjectPermission{Allow: []string{"_INBOX.node.ace.>"}}}}) + } + return out + } + start := func(t *testing.T) (*natsserver.Server, func(logins ...string)) { + opts := &natsserver.Options{Host: "127.0.0.1", Port: natsserver.RANDOM_PORT, NoLog: true, NoSigs: true, Users: users(first)} + srv, err := natsserver.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(20 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + return srv, func(logins ...string) { // the vault's step: the grants part's list, written and loaded by the bus node + next := opts.Clone() + next.Users = users(logins...) + if err := srv.ReloadOptions(next); err != nil { + t.Fatal(err) + } + } + } + dial := func(srv *natsserver.Server, login string) (*nats.Conn, error) { + pw := map[string]string{first: "pw-first", second: "pw-second", "vault": "vault-pw"}[login] + return nats.Connect(srv.ClientURL(), nats.UserInfo(login, pw), nats.CustomInboxPrefix("_INBOX.node.ace"), nats.NoReconnect(), nats.Timeout(3*time.Second)) + } + authenticates := func(srv *natsserver.Server, login string) bool { + c, err := dial(srv, login) + if err == nil { + c.Close() + } + return err == nil + } + oneAuthenticates := func(t *testing.T, srv *natsserver.Server, step string) { + t.Helper() + if !authenticates(srv, first) && !authenticates(srv, second) { + t.Fatalf("%s: no login of the node authenticates", step) + } + } + // The rotator over the real bus: the test-dial and the confirmation are real connections, the confirmation a request to the + // stand-in vault, which names the caller it saw in the subject it was called on. + build := func(t *testing.T, srv *natsserver.Server) (*Rotator, *Logins, chan string) { + confirmed := make(chan string, 4) + vault, err := dial(srv, "vault") + if err != nil { + t.Fatal(err) + } + t.Cleanup(vault.Close) + if _, err := vault.Subscribe("mesh.seat.mesh-vault.call.node-account-confirm.*", func(m *nats.Msg) { + confirmed <- m.Subject[strings.LastIndex(m.Subject, ".")+1:] + _ = m.Respond([]byte(`{"result":{}}`)) + }); err != nil { + t.Fatal(err) + } + _ = vault.Flush() + l := NewLogins("ace", "pw-first", LoginsState{}, nil) + r := &Rotator{Logins: l, FailBound: 2, + Fetch: func(context.Context, string) (string, error) { return "pw-second", nil }, + TestDial: func(_ context.Context, login, pw string) error { + c, err := nats.Connect(srv.ClientURL(), nats.UserInfo(login, pw), nats.CustomInboxPrefix("_INBOX.node.ace"), nats.NoReconnect(), nats.Timeout(3*time.Second)) + if err == nil { + c.Close() + } + return err + }, + Confirm: func(ctx context.Context) error { + login, pw := l.Current() + c, err := nats.Connect(srv.ClientURL(), nats.UserInfo(login, pw), nats.CustomInboxPrefix("_INBOX.node.ace"), nats.NoReconnect(), nats.Timeout(3*time.Second)) + if err != nil { + return err + } + defer c.Close() + reply, err := c.RequestWithContext(ctx, CalledVaultSubjectAs("node-account-confirm", login), []byte(`{}`)) + if err != nil { + return err + } + if _, err := seatAnswer(reply.Data); err != nil { + return errors.New("refused: " + err.Error()) + } + return nil + }} + r.Watch() + return r, l, confirmed + } + ctx := context.Background() + since := time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC) + key := func(login, rotating, stage string) delivery.NodeLogin { + return delivery.NodeLogin{Node: "ace", Login: login, Rotating: rotating, Stage: stage, Since: since} + } + + // A whole rotation: begun, the list holds both logins, ready, the switch and the confirmation, the old login retired, done. + srv, reload := start(t) + r, l, confirmed := build(t, srv) + oneAuthenticates(t, srv, "before the rotation") + r.Heard(ctx, key(first, second, delivery.RotationBegun)) + oneAuthenticates(t, srv, "begun") + reload(first, second) + oneAuthenticates(t, srv, "the list holds both") + r.Heard(ctx, key(first, second, delivery.RotationReady)) + if cur, _ := l.Current(); cur != second { + t.Fatalf("the engine did not switch once the list held both logins: %s", cur) + } + select { + case caller := <-confirmed: + if caller != "node~ace--b" { + t.Errorf("the confirmation came in the name of %s, not the new login's", caller) + } + case <-time.After(5 * time.Second): + t.Fatal("no confirmation reached the vault") + } + oneAuthenticates(t, srv, "switched") + reload(second) // the vault removes the old login's hash after the confirmation + oneAuthenticates(t, srv, "the old login retired") + r.Heard(ctx, key(second, "", "")) + if _, held := l.Value(first); held || authenticates(srv, first) || !authenticates(srv, second) { + t.Errorf("after rotation-done: old value held %v, old authenticates %v, new authenticates %v", held, authenticates(srv, first), authenticates(srv, second)) + } + + // With the list never loaded the engine never switches: the test-dial is refused. + srv, _ = start(t) + r, l, _ = build(t, srv) + r.Heard(ctx, key(first, second, delivery.RotationReady)) + if cur, _ := l.Current(); cur != first || !authenticates(srv, first) { + t.Errorf("the list was never loaded and the engine is on %s", cur) + } + + // An abandon after the switch restores the old login: the vault drops the new one, the engine switches back first. + srv, reload = start(t) + r, l, _ = build(t, srv) + reload(first, second) + r.Heard(ctx, key(first, second, delivery.RotationReady)) + reload(first) + r.Heard(ctx, key(first, "", "")) + if cur, _ := l.Current(); cur != first || !authenticates(srv, first) { + t.Errorf("after an abandon the engine is on %s", cur) + } + if _, held := l.Value(second); held { + t.Error("the abandoned login's value is kept") + } +} diff --git a/internal/link/rotate_nats.go b/internal/link/rotate_nats.go new file mode 100644 index 0000000..b693404 --- /dev/null +++ b/internal/link/rotate_nats.go @@ -0,0 +1,35 @@ +package link + +import ( + "context" + "fmt" + "time" + + "github.com/nats-io/nats.go" +) + +// TestDial dials a second connection to the bus as login with password, on the pinned certificate and the node's one inbox, +// and answers whether it authenticated (novox/hq ADR 0583 §3): the connection is closed at once, nothing is subscribed or said. +func TestDial(ctx context.Context, m Membership, login, password string, timeout time.Duration) error { + config, err := PinnedConfig(m.Fingerprint) + if err != nil { + return err + } + conn, err := nats.Connect(natsURL(m.Broker), + nats.Secure(config), + nats.UserInfo(login, password), + nats.CustomInboxPrefix("_INBOX.node."+m.Node), + nats.Name("mesh-host/"+m.Node+"/test-dial"), + nats.Timeout(timeout), + nats.NoReconnect()) + if err != nil { + return fmt.Errorf("cannot dial as %s: %w", login, err) + } + defer conn.Close() + flush, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + if err := conn.FlushWithContext(flush); err != nil { + return fmt.Errorf("the bus did not answer as %s: %w", login, err) + } + return nil +} diff --git a/internal/link/rotate_test.go b/internal/link/rotate_test.go new file mode 100644 index 0000000..57a8e32 --- /dev/null +++ b/internal/link/rotate_test.go @@ -0,0 +1,159 @@ +package link + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// rotRig is a Rotator over stand-ins: what the vault hands, what a test-dial answers, and what it was asked, in order. +type rotRig struct { + r *Rotator + logins *Logins + states []LoginsState + steps []string + dialErr error + said []string +} + +func newRotRig() *rotRig { + g := &rotRig{} + g.logins = NewLogins("ace", "first", LoginsState{}, func(s LoginsState) error { + g.states = append(g.states, LoginsState{Current: s.Current, Values: map[string]string{}}) + for k, v := range s.Values { + g.states[len(g.states)-1].Values[k] = v + } + return nil + }) + g.r = &Rotator{Logins: g.logins, + Fetch: func(_ context.Context, login string) (string, error) { + g.steps = append(g.steps, "fetch "+login) + return "secret-of-" + login, nil + }, + TestDial: func(_ context.Context, login, _ string) error { + g.steps = append(g.steps, "test-dial "+login) + return g.dialErr + }, + Confirm: func(context.Context) error { + cur, _ := g.logins.Current() + g.steps = append(g.steps, "confirm as "+cur) + return nil + }, + Say: func(s string) { g.said = append(g.said, s) }, + Now: func() time.Time { return time.Date(2026, 11, 2, 12, 5, 0, 0, time.UTC) }, + FailBound: 2} + g.r.Watch() + return g +} + +func key(login, rotating, stage string) delivery.NodeLogin { + return delivery.NodeLogin{Node: "ace", Login: login, Rotating: rotating, Stage: stage, Since: time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC)} +} + +// 774/S62/AC1 and AC2 (ADR 0583 §3): at ready the engine fetches the new value and test-dials it; one that does not +// authenticate leaves it on the old login, saying why; one that does is switched to, confirmed over the new connection, and +// the old value is kept. +func TestTheEngineSwitchesOnlyAfterATestDialAndConfirmsAsTheNewLogin(t *testing.T) { + g := newRotRig() + ctx := context.Background() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationBegun)) + if len(g.steps) != 0 { + t.Fatalf("before ready: %v", g.steps) + } + g.dialErr = errors.New("authorization violation") + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if cur, _ := g.logins.Current(); cur != "node.ace" || !strings.Contains(strings.Join(g.said, "|"), "stays on node.ace") || + slices.Contains(g.steps, "confirm as node.ace--b") { + t.Fatalf("a test-dial that failed: %s, said %v, steps %v", cur, g.said, g.steps) + } + g.dialErr = nil + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + cur, pw := g.logins.Current() + if cur != "node.ace--b" || pw != "secret-of-node.ace--b" { + t.Fatalf("after a good test-dial: %s %s", cur, pw) + } + if want := []string{"fetch node.ace--b", "test-dial node.ace--b", "test-dial node.ace--b", "confirm as node.ace--b"}; !slices.Equal(g.steps, want) { + t.Errorf("steps %v, want %v", g.steps, want) + } + if v, ok := g.logins.Value("node.ace"); !ok || v != "first" { + t.Error("the old value was not kept at the switch") + } + select { + case <-g.logins.Changed(): + default: + t.Error("the link was not told to open again as the new login") + } +} + +// 774/S63/AC1-AC3 (ADR 0583 §4-§6): the old value is dropped at rotation-done and not before; an abandon after the switch, +// before or after the confirmation, switches back before the new value is dropped; repeated failures with the new value before +// rotation-done bring the old one back, said as node-account-fallback, and that rotation is not tried again. +func TestTheOldValueGoesAtRotationDoneAbandonSwitchesBackAndFailuresFallBack(t *testing.T) { + ctx := context.Background() + for _, stage := range []string{"before confirming", "after confirming"} { + g := newRotRig() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if stage == "after confirming" { + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationConfirmed)) + } + if _, held := g.logins.Value("node.ace"); !held { + t.Fatalf("%s: the old value was dropped before rotation-done", stage) + } + g.states = nil + g.r.Heard(ctx, key("node.ace", "", "")) // abandon: the vault keeps the old login in use, no rotation + if cur, _ := g.logins.Current(); cur != "node.ace" { + t.Fatalf("%s: after abandon the engine dials as %s", stage, cur) + } + for _, s := range g.states { + if _, held := s.Values["node.ace--b"]; !held && s.Current != "node.ace" { + t.Errorf("%s: the new value was dropped while %s was in use: %+v", stage, s.Current, s) + } + } + if _, held := g.logins.Value("node.ace--b"); held { + t.Errorf("%s: the abandoned value is kept", stage) + } + } + g := newRotRig() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + g.r.Heard(ctx, key("node.ace--b", "", "")) // rotation-done: the vault retired the old login + if _, held := g.logins.Value("node.ace"); held { + t.Error("the old value is kept after rotation-done") + } + if cur, _ := g.logins.Current(); cur != "node.ace--b" { + t.Errorf("in use after done: %s", cur) + } + // Fallback: two failures with the new value (the bound) before rotation-done. + f := newRotRig() + f.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + f.logins.dialed("node.ace--b", errors.New("nats: authorization violation")) + if cur, _ := f.logins.Current(); cur != "node.ace--b" { + t.Fatalf("fell back after one failure: %s", cur) + } + f.logins.dialed("node.ace--b", errors.New("nats: authorization violation")) + fb := f.r.Fallback() + if cur, _ := f.logins.Current(); cur != "node.ace" || fb == nil || fb.Node != "ace" || fb.From != "node.ace--b" || fb.Why == "" || fb.At.IsZero() { + t.Fatalf("after the bound: %s, fall back %+v", cur, fb) + } + // The health report carries it, from which the watchman raises node-account-fallback. + if _, err := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: "ace", Fallback: fb}); err != nil { + t.Errorf("the fall back is no health report's: %v", err) + } + steps := len(f.steps) + f.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if len(f.steps) != steps { + t.Errorf("the rotation fallen back from was tried again: %v", f.steps[steps:]) + } + f.logins.dialed("node.ace", errors.New("dial tcp: connection refused")) // not a refusal of credentials: counts for nothing + // The vault moves on (it abandons the rotation): the fall back is said no more. + abandoned := key("node.ace", "", "") + abandoned.Since = abandoned.Since.Add(time.Minute) + f.r.Heard(ctx, abandoned) + if f.r.Fallback() != nil { + t.Error("the fall back is still said after the vault moved on") + } +} diff --git a/internal/link/run.go b/internal/link/run.go index 70c441c..968d504 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -21,6 +21,9 @@ type Membership struct { // Transport is which bus this membership was minted for (hearing.go): the mesh's own, and a // membership that names another is one this host cannot dial with. Transport string + // Logins is which of its two logins the node dials as (ADR 0583 §1, logins.go). Nil is node. with Password, as before + // any rotation. + Logins *Logins } // Announce is how the link says what is happening, so a node running unattended leaves an @@ -240,6 +243,10 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann // seat, the hand-over and every other subject heard here stayed deaf until the engine restarted (the sweep of // 2026-10-30: every engine was refused its node seat at 12:10 and answered nobody all afternoon). The link is kept // for a pause, said, and then dropped, so Hold opens it again and every subscription is asked anew. + var loginChanged <-chan struct{} + if m.Logins != nil { + loginChanged = m.Logins.Changed() + } var refused <-chan string if r, ok := link.(Refusals); ok { refused = r.Refused() @@ -259,6 +266,8 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann return nil case reason := <-link.Lost(): return reason + case <-loginChanged: + return errors.New("the login this node dials as changed (ADR 0583): the link is opened again as it") case <-asks: queue.FactsAsked(ctx) case subject := <-refused: diff --git a/internal/link/session.go b/internal/link/session.go index 5f63d1f..832348e 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/delivery" "sync" "time" @@ -24,6 +25,10 @@ const Writer = "node-engine" // event of the node, ordered by the node-engine's own epoch and sequence. Nothing here receives, orders or reconciles // a whole declaration. type Session struct { + // Fallback is the engine's fall back to its old login its health report carries now (Rotator.Fallback); nil says none. + Fallback func() *delivery.LoginFallback + // linkedUser is the bus user the link open now authenticated as, guarded by mu (WhenLinkedAs). + linkedUser string // Membership is the node and the mesh's key a controller's ask is verified with. Membership Membership // OnLink is run, off the link's own loop, each time a link opens and when the controller asks the node to say its @@ -197,6 +202,10 @@ func (q *Session) Attach(ctx context.Context, bus Bus) { q.init() q.mu.Lock() q.bus = bus + q.linkedUser = "" + if u, ok := bus.(interface{ User() string }); ok { + q.linkedUser = u.User() + } q.gone = make(chan struct{}) close(q.changed) q.changed = make(chan struct{}) @@ -224,6 +233,7 @@ func (q *Session) Detach(bus Bus) { q.mu.Lock() if q.bus == bus { q.bus = nil + q.linkedUser = "" close(q.gone) close(q.changed) q.changed = make(chan struct{}) diff --git a/internal/link/watch.go b/internal/link/watch.go index 145d5aa..9f03840 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -196,6 +196,26 @@ func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.Canc } } +// WhenLinkedAs waits for a link open that authenticated as user: after a switch of login, the vault is asked over the new +// connection and no other (ADR 0583 §3). +func (q *Session) WhenLinkedAs(ctx context.Context, user string) error { + q.init() + for { + q.mu.Lock() + if q.bus != nil && q.linkedUser == user { + q.mu.Unlock() + return nil + } + changed := q.changed + q.mu.Unlock() + select { + case <-ctx.Done(): + return ctx.Err() + case <-changed: + } + } +} + // EventBus is a link that can hear an event on a subject of core NATS: the node agent waits for a backup's end with it, // and never polls. type EventBus interface {