The engine dials as the login node-logins names, rotates it after a test-dial, and enrols through the enrolment seat (issue 774) #195
@@ -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
|
||||
}
|
||||
@@ -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/<node>/mesh-host/<login> (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
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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")
|
||||
|
||||
+10
-107
@@ -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) }
|
||||
|
||||
@@ -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.<node>
|
||||
// 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.<node>, 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.<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() }
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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.<n> or node.<n>--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.<name>" (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 }
|
||||
|
||||
@@ -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.<n> and node.<n>--b with the node's permissions and its inbox
|
||||
// _INBOX.node.<n> 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.<n> 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.<n>) 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.<n> (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.<n> and the membership's password otherwise (the inbox, _INBOX.node.<n>, 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")
|
||||
}
|
||||
@@ -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.<n>
|
||||
// with the membership's password as before; with a key naming node.<n>--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()))
|
||||
}
|
||||
}
|
||||
@@ -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.<n> or node.<n>--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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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/<node>/mesh-host/<login> (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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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.<n> 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:
|
||||
|
||||
@@ -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{})
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user