diff --git a/cmd/mesh-host/logins.go b/cmd/mesh-host/logins.go new file mode 100644 index 0000000..574a3fd --- /dev/null +++ b/cmd/mesh-host/logins.go @@ -0,0 +1,86 @@ +package main + +import ( + "context" + "encoding/json" + "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() + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 97d0bc6..8b656eb 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" @@ -1074,6 +1075,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 +1193,12 @@ 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). + go followLogins(aside, queue, mine.Node, func(nl delivery.NodeLogin) { + if line := membership.Logins.Heard(nl); line != "" { + say(line) + } + }, 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/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index afc81b4..0ba79dc 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -77,14 +77,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), diff --git a/internal/link/logins.go b/internal/link/logins.go new file mode 100644 index 0000000..8cb08be --- /dev/null +++ b/internal/link/logins.go @@ -0,0 +1,181 @@ +package link + +import ( + "sync" + + "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{} +} + +// 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 +} 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/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/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: