Merge pull request 'Bus node: follow the grants part's sealed user list after the operator's take-over, never both sources; hand the vault the loaded hashes (issue 1945)' (#145) from feat/1945-user-list-switch into main
This commit was merged in pull request #145.
This commit is contained in:
@@ -1153,7 +1153,8 @@ func runLink(ctx context.Context, opts options) error {
|
||||
return identity.LoadSealingKey(identity.SealingKeyPath(opts.state))
|
||||
}, func(verb string, args any) (json.RawMessage, error) {
|
||||
return queue.AskVault(ctx, verb, args)
|
||||
}, say, nodeKeyRetryFrom, nodeKeyRetryUpTo)
|
||||
}, say, nodeKeyRetryFrom, nodeKeyRetryUpTo, &keyHearing{heard: vaultKey.heard,
|
||||
held: func() bool { return vaultKey.get() != "" }, look: nodeKeyLook, every: nodeKeyReask})
|
||||
// And whether what it runs stays up, looked at on its own clock and said when it changes (novox/hq
|
||||
// ADR 0240) — between applies, which is when a container crash-loops.
|
||||
if judging != nil {
|
||||
|
||||
@@ -412,6 +412,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
}
|
||||
})
|
||||
})
|
||||
// A vault key heard (or heard again changed) looks at the list again, so its hashes are handed (ADR 0547 §9).
|
||||
vaultKey.following(users.Again)
|
||||
go func() {
|
||||
if err := users.Run(linked); err != nil && linked.Err() == nil {
|
||||
say("the bus's user list is not followed on this link: " + err.Error())
|
||||
@@ -447,7 +449,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
Backoff: link.Backoff,
|
||||
Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
|
||||
actx, done := applyContext(session.WithLink(ctx))
|
||||
res := applyComposition(actx, opts, c, others, sched, say)
|
||||
// While the grants part's list is followed, the controller's list in the bus module's composition is not
|
||||
// written: the apply keeps the list on the machine (ADR 0547 clause 8, never both).
|
||||
held, herr := users.Hold(c)
|
||||
var res agent.Result
|
||||
if herr != nil {
|
||||
res = failedWith(herr.Error())
|
||||
} else {
|
||||
res = applyComposition(actx, opts, held, others, sched, say)
|
||||
}
|
||||
done()
|
||||
// An apply of the bus module may write an older list than the newest composition holds: looked at again.
|
||||
if module, _ := users.State(); module != "" && module == c.Module {
|
||||
|
||||
@@ -2,7 +2,10 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/identity"
|
||||
@@ -22,12 +25,77 @@ const (
|
||||
nodeKeyLook = 30 * time.Second
|
||||
)
|
||||
|
||||
// heardKey is the vault's own public key as its answer to node-key gave it (ADR 0547 clause 9, issue 1945): the key the
|
||||
// bus node hands the loaded list's hashes sealed to. One for the process, as the key stated is; kick is the follower of
|
||||
// the user list on the link open now, told when the key is first heard or changes.
|
||||
type heardKey struct {
|
||||
mu sync.Mutex
|
||||
pem string
|
||||
kick func()
|
||||
}
|
||||
|
||||
// vaultKey is the vault's key this engine heard.
|
||||
var vaultKey = &heardKey{}
|
||||
|
||||
func (h *heardKey) get() string {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return h.pem
|
||||
}
|
||||
|
||||
// heard keeps the key a node-key answer gave, and tells the follower when it is a first or changed one. An answer
|
||||
// with no key (a vault before this one) changes nothing.
|
||||
func (h *heardKey) heard(answer json.RawMessage) {
|
||||
var a struct {
|
||||
VaultKey string `json:"vault_key"`
|
||||
}
|
||||
if json.Unmarshal(answer, &a) != nil || !strings.Contains(a.VaultKey, "PUBLIC KEY") {
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
changed := a.VaultKey != h.pem
|
||||
h.pem = a.VaultKey
|
||||
kick := h.kick
|
||||
h.mu.Unlock()
|
||||
if changed && kick != nil {
|
||||
kick()
|
||||
}
|
||||
}
|
||||
|
||||
// following names the follower told when the key changes, and tells it at once when a key is known.
|
||||
func (h *heardKey) following(kick func()) {
|
||||
h.mu.Lock()
|
||||
h.kick = kick
|
||||
known := h.pem != ""
|
||||
h.mu.Unlock()
|
||||
if known && kick != nil {
|
||||
kick()
|
||||
}
|
||||
}
|
||||
|
||||
// nodeKeyReask is how often the key is stated again though it did not change, so a changed vault key is heard: the
|
||||
// vault takes the same key again and changes nothing (ADR 0547 clause 9).
|
||||
const nodeKeyReask = 10 * time.Minute
|
||||
|
||||
// keyHearing is what the keeper does with the vault's own key in node-key's answers (ADR 0547 clause 9): heard takes
|
||||
// it from an answer; held says whether one is held, and while none is the key is stated again at every look, so a vault
|
||||
// delivered after this engine is heard within a look; every is how often it is stated again all the same, so a changed
|
||||
// vault key is heard.
|
||||
type keyHearing struct {
|
||||
heard func(json.RawMessage)
|
||||
held func() bool
|
||||
// look is how often it is stated again while no vault key is held; every how often once one is.
|
||||
look, every time.Duration
|
||||
}
|
||||
|
||||
// keepNodeKeyStated states the sealing key load answers to the vault at start, retrying with a growing wait from
|
||||
// retryFrom to retryUpTo while it does not answer, and states it again whenever the key changes. It says one line
|
||||
// when a key is stated and one when a try fails for another reason than the last, and returns when ctx ends.
|
||||
// when a key is stated and one when a try fails for another reason than the last, and returns when ctx ends. With
|
||||
// hearing, each answer goes to it, and the key is stated again while no vault key is held and at its interval.
|
||||
func keepNodeKeyStated(ctx context.Context, load func() (identity.SealingKey, error), ask vaultAsk, say func(string),
|
||||
retryFrom, retryUpTo time.Duration) {
|
||||
retryFrom, retryUpTo time.Duration, hearing *keyHearing) {
|
||||
stated, lastFailure := "", ""
|
||||
var askedAt time.Time
|
||||
wait := retryFrom
|
||||
fail := func(why string) {
|
||||
if why != lastFailure {
|
||||
@@ -40,17 +108,29 @@ func keepNodeKeyStated(ctx context.Context, load func() (identity.SealingKey, er
|
||||
for {
|
||||
delay := look
|
||||
key, err := load()
|
||||
again := false
|
||||
if hearing != nil && stated != "" {
|
||||
since := time.Since(askedAt)
|
||||
held := hearing.held == nil || hearing.held()
|
||||
again = (!held && since >= hearing.look) || (held && hearing.every > 0 && since >= hearing.every)
|
||||
}
|
||||
switch {
|
||||
case err != nil:
|
||||
fail(err.Error())
|
||||
delay = wait
|
||||
case key.Public != stated:
|
||||
if _, err := ask("node-key", map[string]string{"key": key.Public}); err != nil {
|
||||
case key.Public != stated || again:
|
||||
if answer, err := ask("node-key", map[string]string{"key": key.Public}); err != nil {
|
||||
fail(err.Error())
|
||||
delay = wait
|
||||
} else {
|
||||
askedAt = time.Now()
|
||||
if hearing != nil && hearing.heard != nil {
|
||||
hearing.heard(answer)
|
||||
}
|
||||
if key.Public != stated {
|
||||
say(fmt.Sprintf("mesh-host: this machine's sealing key %s is stated to the vault", key.Public))
|
||||
}
|
||||
stated, lastFailure, wait = key.Public, "", retryFrom
|
||||
say(fmt.Sprintf("mesh-host: this machine's sealing key %s is stated to the vault", key.Public))
|
||||
}
|
||||
}
|
||||
select {
|
||||
|
||||
@@ -19,6 +19,10 @@ type keyVault struct {
|
||||
stated []string
|
||||
fail int
|
||||
verbs []string
|
||||
// keyless is how many answers carry no vault key (a vault before ADR 0547 clause 9's, or not yet started); key the
|
||||
// vault's key the others carry.
|
||||
keyless int
|
||||
key string
|
||||
}
|
||||
|
||||
func (v *keyVault) ask(verb string, args any) (json.RawMessage, error) {
|
||||
@@ -35,7 +39,12 @@ func (v *keyVault) ask(verb string, args any) (json.RawMessage, error) {
|
||||
}
|
||||
_ = json.Unmarshal(raw, &a)
|
||||
v.stated = append(v.stated, a.Key)
|
||||
return json.RawMessage(`{"recorded":true}`), nil
|
||||
if v.key == "" || v.keyless > 0 {
|
||||
v.keyless--
|
||||
return json.RawMessage(`{"recorded":true}`), nil
|
||||
}
|
||||
answer, _ := json.Marshal(map[string]any{"recorded": true, "vault_key": v.key})
|
||||
return answer, nil
|
||||
}
|
||||
|
||||
func (v *keyVault) said() []string {
|
||||
@@ -71,9 +80,14 @@ func TestTheEngineStatesItsSealingKeyToTheVaultAtStartRetriesAndRestatesAChange(
|
||||
say := func(l string) { lmu.Lock(); lines = append(lines, l); lmu.Unlock() }
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go keepNodeKeyStated(ctx, load, vault.ask, say, time.Millisecond, 5*time.Millisecond)
|
||||
var hmu sync.Mutex
|
||||
var answers []string
|
||||
heard := func(a json.RawMessage) { hmu.Lock(); answers = append(answers, string(a)); hmu.Unlock() }
|
||||
go keepNodeKeyStated(ctx, load, vault.ask, say, time.Millisecond, 5*time.Millisecond, &keyHearing{heard: heard})
|
||||
|
||||
waitFor(t, "the first key stated after two refusals", func() bool { return len(vault.said()) == 1 })
|
||||
// Its answer is handed on, so the vault's own key in it is heard (ADR 0547 clause 9); refusals hand nothing.
|
||||
waitFor(t, "the answer heard", func() bool { hmu.Lock(); defer hmu.Unlock(); return len(answers) == 1 })
|
||||
if vault.said()[0] != first.Public {
|
||||
t.Errorf("stated %q, want the engine's own key %q", vault.said()[0], first.Public)
|
||||
}
|
||||
@@ -102,11 +116,75 @@ func TestTheEngineStatesItsSealingKeyToTheVaultAtStartRetriesAndRestatesAChange(
|
||||
if len(lines) == 0 {
|
||||
t.Error("the engine said nothing")
|
||||
}
|
||||
asksAgainUntilItHearsTheVaultsKey(t)
|
||||
}
|
||||
|
||||
// The statement a host makes names this contract (issue 786), one above the vault's box.
|
||||
// ADR 0547 clause 9: the vault's key is heard though the engine's own key never changes. While none is held the key
|
||||
// is stated again at each look, so a vault delivered after the engine (or answering no key yet) is heard; once one is
|
||||
// held, again only at the slow interval, so a changed vault key is heard; never more.
|
||||
func asksAgainUntilItHearsTheVaultsKey(t *testing.T) {
|
||||
sealing, _ := identity.GenerateSealingKey()
|
||||
load := func() (identity.SealingKey, error) { return sealing, nil }
|
||||
pem := func(c string) string { return "-----BEGIN PUBLIC KEY-----\n" + c + "\n-----END PUBLIC KEY-----\n" }
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
keyless int
|
||||
every time.Duration
|
||||
change bool
|
||||
wantAtLeast int
|
||||
wantAtMost int
|
||||
}{
|
||||
{name: "a vault with no key yet is asked again until it answers one, then not", keyless: 3, wantAtLeast: 4, wantAtMost: 4},
|
||||
{name: "a held key is asked again at the slow interval, and a changed one heard", every: 20 * time.Millisecond,
|
||||
change: true, wantAtLeast: 3, wantAtMost: 1 << 20},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
vault := &keyVault{keyless: c.keyless, key: pem("AAAA")}
|
||||
h := &heardKey{}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go keepNodeKeyStated(ctx, load, vault.ask, func(string) {}, time.Millisecond, 5*time.Millisecond,
|
||||
&keyHearing{heard: h.heard, held: func() bool { return h.get() != "" }, look: time.Millisecond, every: c.every})
|
||||
waitFor(t, "the vault's key heard", func() bool { return h.get() == pem("AAAA") })
|
||||
if c.change {
|
||||
vault.mu.Lock()
|
||||
vault.key = pem("BBBB")
|
||||
vault.mu.Unlock()
|
||||
waitFor(t, "the changed vault key heard", func() bool { return h.get() == pem("BBBB") })
|
||||
}
|
||||
time.Sleep(60 * time.Millisecond)
|
||||
n := len(vault.said())
|
||||
if n < c.wantAtLeast || n > c.wantAtMost {
|
||||
t.Errorf("node-key asked %d times, want %d to %d", n, c.wantAtLeast, c.wantAtMost)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The statement a host makes names this contract (issue 786), one above the vault's box. And the vault's key its
|
||||
// answer gives is kept, told to the follower once when first heard and again only when changed; an answer without one
|
||||
// (an older vault) or with no public key changes nothing (ADR 0547 clause 9).
|
||||
func TestTheNodeKeyContractIsThirteen(t *testing.T) {
|
||||
if link.NodeKeyContract != 13 {
|
||||
t.Errorf("NodeKeyContract %d", link.NodeKeyContract)
|
||||
}
|
||||
h := &heardKey{}
|
||||
kicks := 0
|
||||
h.following(func() { kicks++ })
|
||||
h.heard(json.RawMessage(`{"recorded":true}`))
|
||||
h.heard(json.RawMessage(`{"recorded":true,"vault_key":"not a key"}`))
|
||||
if h.get() != "" || kicks != 0 {
|
||||
t.Fatalf("an answer with no key was kept: %q, %d kicks", h.get(), kicks)
|
||||
}
|
||||
pem := "-----BEGIN PUBLIC KEY-----\nAAAA\n-----END PUBLIC KEY-----\n"
|
||||
answer, _ := json.Marshal(map[string]any{"recorded": true, "vault_key": pem})
|
||||
h.heard(answer)
|
||||
h.heard(answer)
|
||||
if h.get() != pem || kicks != 1 {
|
||||
t.Fatalf("key %q, %d kicks, want the key and one", h.get(), kicks)
|
||||
}
|
||||
h.following(func() { kicks += 10 })
|
||||
if kicks != 11 {
|
||||
t.Fatalf("a follower named after the key was heard was not told: %d", kicks)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,9 +7,13 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/seal"
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/identity"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
"github.com/novox/mesh-host/internal/system"
|
||||
@@ -42,10 +46,45 @@ func userListFollower(opts options, session *link.Session, composedBucket string
|
||||
Report: func(ctx context.Context, module string, w userlist.Written) {
|
||||
session.NoteUserList(ctx, module, w)
|
||||
},
|
||||
Say: announceOr(say),
|
||||
// From take-over, the grants part's list (ADR 0547 clauses 7, 8): its one key watched, its object read by name and
|
||||
// opened with this machine's sealing key, the key's public half said in the have so the grants part seals to it.
|
||||
Grants: bucketWatch{session, delivery.GrantsUserListBucket},
|
||||
Fetch: func(ctx context.Context, name string) ([]byte, error) {
|
||||
return session.GrantsObject(ctx, name, grantsListLimit)
|
||||
},
|
||||
Open: func(b seal.Box) ([]byte, error) {
|
||||
key, err := identity.LoadSealingKey(identity.SealingKeyPath(opts.state))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return key.OpenVaultBox(b)
|
||||
},
|
||||
Key: func() string {
|
||||
key, err := identity.LoadSealingKey(identity.SealingKeyPath(opts.state))
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
pem, err := key.PublicPEM()
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return pem
|
||||
},
|
||||
// The grants part's key as last heard, kept beside the node's state, followed while it cannot be watched; its
|
||||
// watch tried again as a passing failure of the bus is (ADR 0547 clause 8).
|
||||
Kept: func() ([]byte, uint64) { return keptGrantsKey(opts) },
|
||||
Keep: func(value []byte, rev uint64) { keepGrantsKey(opts, value, rev, say) },
|
||||
Backoff: link.Backoff,
|
||||
// The vault's key from its answer to node-key: the loaded list's hashes are handed sealed to it (ADR 0547
|
||||
// clause 9).
|
||||
VaultKey: vaultKey.get,
|
||||
Say: announceOr(say),
|
||||
})
|
||||
}
|
||||
|
||||
// grantsListLimit is the largest sealed list the bus node reads: the live list is about a hundred kilobytes.
|
||||
const grantsListLimit = 16 << 20
|
||||
|
||||
// userListFile is a user-list resource as the file it declares.
|
||||
func userListFile(raw []byte) (*declaration.File, error) {
|
||||
r, err := declaration.ParseResource(raw)
|
||||
@@ -122,3 +161,44 @@ func writeUserListAlone(ctx context.Context, opts options, raw []byte, body []by
|
||||
}
|
||||
return digest, writeErr == nil && out.Action != "unchanged", writeErr
|
||||
}
|
||||
|
||||
// grantsKeyPath is where the grants part's user-list key last heard is kept, beside the node's state.
|
||||
func grantsKeyPath(opts options) string {
|
||||
return filepath.Join(filepath.Dir(opts.state), "grants-user-list.json")
|
||||
}
|
||||
|
||||
type keptKey struct {
|
||||
Revision uint64 `json:"revision"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
}
|
||||
|
||||
func keptGrantsKey(opts options) ([]byte, uint64) {
|
||||
raw, err := os.ReadFile(grantsKeyPath(opts))
|
||||
if err != nil {
|
||||
return nil, 0
|
||||
}
|
||||
var k keptKey
|
||||
if json.Unmarshal(raw, &k) != nil || len(k.Value) == 0 {
|
||||
return nil, 0
|
||||
}
|
||||
return k.Value, k.Revision
|
||||
}
|
||||
|
||||
func keepGrantsKey(opts options, value []byte, rev uint64, say link.Announce) {
|
||||
path := grantsKeyPath(opts)
|
||||
if value == nil {
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
announceOr(say)("the grants part's user-list key kept on this machine could not be taken away: " + err.Error())
|
||||
}
|
||||
return
|
||||
}
|
||||
raw, _ := json.Marshal(keptKey{Revision: rev, Value: value})
|
||||
tmp := path + ".new"
|
||||
err := os.WriteFile(tmp, raw, 0o600)
|
||||
if err == nil {
|
||||
err = os.Rename(tmp, path)
|
||||
}
|
||||
if err != nil {
|
||||
announceOr(say)("the grants part's user-list key could not be kept on this machine: " + err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.20261025204730-0be0ffd6159e
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376
|
||||
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
|
||||
|
||||
@@ -4,6 +4,14 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456 h1:He2o9vSK
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e h1:k/1NjjW5HuuuHW2ldyWTvjp+1u3fN0RO18Q57Iup6Dk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026105311-175a5567b8bd h1:ef6y09csev6xY/suFrIFGUXbgvDA29kummFtHcIb3B8=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026105311-175a5567b8bd/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026110029-83ee7769d8ee h1:ZdBMkF4WYJi82jb3xYi9Z2lXN/iMTn9r3SFC122ROY4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026110029-83ee7769d8ee/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112307-af66adc180cd h1:DV4xfPRqmh6p20D2QNN+9EGFAyvpGOZPTRsSbpT7oww=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112307-af66adc180cd/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376 h1:by2zBASLb8h5jA1IiYJWykXwLnHYpuZh/zFNa4NsbOU=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026112527-c2a410bb6376/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=
|
||||
|
||||
+50
-2
@@ -10,6 +10,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
@@ -92,7 +93,32 @@ func (b OverNATS) FileObject(ctx context.Context, name string, limit int64) ([]b
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stream := declaration.FileStreamPrefix + bucket
|
||||
return objectByName(ctx, js, bucket, declaration.FileStreamPrefix+bucket, name, limit)
|
||||
}
|
||||
|
||||
// GrantsObject reads one object of the grants part's bucket of sealed user lists by its name (novox/hq ADR 0547 clause 7),
|
||||
// with the same subjects and nothing more as FileObject: the bus node is granted exactly those (mesh-sdk
|
||||
// buscompose.GrantsUserListFollows).
|
||||
func (b OverNATS) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
||||
js := b.JS
|
||||
wait := 10 * time.Second
|
||||
if at, ok := ctx.Deadline(); ok {
|
||||
if wait = time.Until(at); wait <= 0 {
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
}
|
||||
if b.Conn != nil {
|
||||
bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
js = bounded
|
||||
}
|
||||
return objectByName(ctx, js, delivery.GrantsUserListObjects, delivery.GrantsUserListObjectStream, name, limit)
|
||||
}
|
||||
|
||||
// objectByName is one object of bucket (its stream stream) by its name, no larger than limit, verified against its digest.
|
||||
func objectByName(ctx context.Context, js nats.JetStreamContext, bucket, stream, name string, limit int64) ([]byte, error) {
|
||||
meta, err := js.GetLastMsg(stream, "$O."+bucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)),
|
||||
nats.DirectGet(), nats.Context(ctx))
|
||||
if err != nil {
|
||||
@@ -109,7 +135,7 @@ func (b OverNATS) FileObject(ctx context.Context, name string, limit int64) ([]b
|
||||
return nil, fmt.Errorf("the object %s was deleted from the bucket %s", name, bucket)
|
||||
}
|
||||
if info.Size > uint64(limit) {
|
||||
return nil, fmt.Errorf("the object is larger than the %d bytes the declaration names", limit)
|
||||
return nil, fmt.Errorf("the object is larger than the %d bytes allowed", limit)
|
||||
}
|
||||
if info.Size == 0 {
|
||||
return []byte{}, nil
|
||||
@@ -142,6 +168,28 @@ func (l *natsLink) FileObject(ctx context.Context, name string, limit int64) ([]
|
||||
return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.FileObject(ctx, name, limit)
|
||||
}
|
||||
|
||||
func (l *natsLink) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
||||
return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.GrantsObject(ctx, name, limit)
|
||||
}
|
||||
|
||||
// GrantsObjects is a link that can read an object of the grants part's bucket of sealed user lists.
|
||||
type GrantsObjects interface {
|
||||
GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error)
|
||||
}
|
||||
|
||||
// GrantsObject reads an object of the grants part's bucket by its name on the link open now.
|
||||
func (q *Session) GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
g, ok := bus.(GrantsObjects)
|
||||
if bus == nil || !ok {
|
||||
return nil, ErrNoLink
|
||||
}
|
||||
return g.GrantsObject(ctx, name, limit)
|
||||
}
|
||||
|
||||
var (
|
||||
_ FilesBus = OverNATS{}
|
||||
_ FilesBus = (*natsLink)(nil)
|
||||
|
||||
@@ -15,6 +15,18 @@
|
||||
// older list, and when asked (Again).
|
||||
//
|
||||
// It reads no want and writes none, moves no fingerprint, and touches nothing of the bus module but its user list.
|
||||
//
|
||||
// **It hands the loaded list's hashes to the vault** (ADR 0547 clause 9, issue 1945): once the vault's key is known
|
||||
// (VaultKey), the have of the bus module carries the bcrypt hash of every user of the list on the machine, sealed to
|
||||
// that key (delivery.AccountsHanded), so the vault adopts the accounts it holds no value for. It is sealed again only
|
||||
// when the list on the machine or the vault's key changes.
|
||||
//
|
||||
// **From take-over it follows the grants part's list instead** (ADR 0547 clauses 7 and 8, issue 1945). On the node whose
|
||||
// compositions carry the list, it also watches the grants part's one key `user-list.sealed`. While that key says
|
||||
// FollowGrants, the list is the object the key names, read by its name, opened with this node's sealing key and checked
|
||||
// against the key's digest, and written to the same file (mode 0600) only when the file's digest differs; otherwise the
|
||||
// controller's compositions are the source, as before. Never both. The have says which it follows and this node's key,
|
||||
// which the grants part seals to.
|
||||
package userlist
|
||||
|
||||
import (
|
||||
@@ -31,6 +43,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/seal"
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
)
|
||||
|
||||
@@ -59,8 +72,24 @@ type Env struct {
|
||||
Write func(ctx context.Context, module string, resource json.RawMessage, body []byte) (digest string, changed bool, err error)
|
||||
// Report says what was written, in the module's have.
|
||||
Report func(ctx context.Context, module string, w Written)
|
||||
Now func() time.Time
|
||||
Say func(string)
|
||||
// Grants watches the grants part's state (its key delivery.GrantsUserListKey); nil follows the controller's alone.
|
||||
Grants agent.Watcher
|
||||
// Fetch reads an object of the grants part's bucket by its name; Open opens a box sealed to this node's key; Key is
|
||||
// this node's sealing key's public half (SPKI PEM), said in the have so the grants part seals to it ("" unknown).
|
||||
Fetch func(ctx context.Context, name string) ([]byte, error)
|
||||
Open func(seal.Box) ([]byte, error)
|
||||
Key func() string
|
||||
// Kept is the grants part's key as last heard on this machine and its revision (nil when none), and Keep keeps it:
|
||||
// what the follower follows while the key cannot be watched (ADR 0547 clause 8). Backoff is the wait before the
|
||||
// try-th watch again.
|
||||
Kept func() ([]byte, uint64)
|
||||
Keep func(value []byte, rev uint64)
|
||||
Backoff func(try int) time.Duration
|
||||
// VaultKey is the vault's own public key (SPKI PEM) as its answer to node-key gave it, "" while none is known: the
|
||||
// key the loaded list's hashes are handed sealed to. Nil hands nothing.
|
||||
VaultKey func() string
|
||||
Now func() time.Time
|
||||
Say func(string)
|
||||
}
|
||||
|
||||
type source struct {
|
||||
@@ -83,6 +112,13 @@ type Follower struct {
|
||||
state Written
|
||||
mod string
|
||||
said map[string]bool
|
||||
// grants is the grants part's key as last heard (nil when none), and following the source last taken: FollowGrants
|
||||
// or FollowController.
|
||||
grants *delivery.GrantsUserList
|
||||
grantsRev uint64
|
||||
following string
|
||||
// unwatched is why the grants part's key is not watched now; empty while it is, or where it need not be.
|
||||
unwatched string
|
||||
}
|
||||
|
||||
// New is a follower for a node.
|
||||
@@ -93,7 +129,18 @@ func New(env Env) *Follower {
|
||||
if env.Say == nil {
|
||||
env.Say = func(string) {}
|
||||
}
|
||||
return &Follower{env: env, kick: make(chan struct{}, 1), srcs: map[string]source{}, said: map[string]bool{}}
|
||||
if env.Backoff == nil {
|
||||
env.Backoff = func(int) time.Duration { return time.Minute }
|
||||
}
|
||||
f := &Follower{env: env, kick: make(chan struct{}, 1), srcs: map[string]source{}, said: map[string]bool{}}
|
||||
if env.Grants != nil && env.Kept != nil {
|
||||
if raw, rev := env.Kept(); raw != nil {
|
||||
if g, err := delivery.DecodeGrantsUserList(raw); err == nil {
|
||||
f.grants, f.grantsRev = &g, rev
|
||||
}
|
||||
}
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// Again asks the follower to look at the file again, though no composition changed: after an apply of the bus module,
|
||||
@@ -120,6 +167,33 @@ func (f *Follower) Run(ctx context.Context) error {
|
||||
return fmt.Errorf("the user list cannot be followed: %w", err)
|
||||
}
|
||||
caught := false
|
||||
// The grants part's key, on the node whose compositions carry the list: the bus node alone is granted it (mesh-sdk
|
||||
// buscompose.GrantsUserListFollows). Until its watch caught up nothing is written, so a new follower (one per link)
|
||||
// never writes the controller's list after take-over before it heard the switch; a watch refused or ended is tried
|
||||
// again on the backoff, and meanwhile the follow last heard (kept on the machine) holds.
|
||||
var grants <-chan agent.Update
|
||||
var retry <-chan time.Time
|
||||
opened, grantsCaught, tries := false, false, 0
|
||||
failed := func(why string) {
|
||||
grants, tries = nil, tries+1
|
||||
retry = time.After(f.env.Backoff(tries))
|
||||
f.mu.Lock()
|
||||
f.unwatched = why
|
||||
f.mu.Unlock()
|
||||
f.sayOnce("the grants part's user-list key is not watched, tried again: " + why)
|
||||
}
|
||||
open := func() {
|
||||
if grants != nil || retry != nil || f.env.Grants == nil || !f.busNode() {
|
||||
return
|
||||
}
|
||||
opened = true
|
||||
ch, err := f.env.Grants.Watch(ctx, delivery.GrantsUserListKey)
|
||||
if err != nil {
|
||||
failed(err.Error())
|
||||
return
|
||||
}
|
||||
grants = ch
|
||||
}
|
||||
for {
|
||||
again := false
|
||||
select {
|
||||
@@ -127,6 +201,8 @@ func (f *Follower) Run(ctx context.Context) error {
|
||||
return nil
|
||||
case <-f.kick:
|
||||
again = true
|
||||
case <-retry:
|
||||
retry = nil
|
||||
case u, ok := <-updates:
|
||||
if !ok {
|
||||
return fmt.Errorf("the watch of the compositions ended")
|
||||
@@ -136,13 +212,99 @@ func (f *Follower) Run(ctx context.Context) error {
|
||||
} else {
|
||||
f.take(u)
|
||||
}
|
||||
case u, ok := <-grants:
|
||||
if !ok {
|
||||
failed("its watch ended")
|
||||
break
|
||||
}
|
||||
if u.Caught {
|
||||
grantsCaught, tries = true, 0
|
||||
f.mu.Lock()
|
||||
f.unwatched = ""
|
||||
f.mu.Unlock()
|
||||
break
|
||||
}
|
||||
f.takeGrants(u)
|
||||
}
|
||||
if caught {
|
||||
f.settle(ctx, again)
|
||||
if !caught {
|
||||
continue
|
||||
}
|
||||
open()
|
||||
if opened && !grantsCaught && grants != nil {
|
||||
continue // the switch is not heard yet: nothing is written before it is
|
||||
}
|
||||
f.mu.Lock()
|
||||
if f.unwatched != f.state.Unwatched {
|
||||
again = true // the have says whether the key is watched
|
||||
}
|
||||
f.mu.Unlock()
|
||||
f.settle(ctx, again)
|
||||
}
|
||||
}
|
||||
|
||||
// busNode says some composition of this node carries the bus's user list.
|
||||
func (f *Follower) busNode() bool {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.cur.key != "" || len(f.srcs) > 0
|
||||
}
|
||||
|
||||
// takeGrants keeps the grants part's key as heard; a value it cannot read is said and leaves the controller's list
|
||||
// followed (no list is followed that the node cannot check).
|
||||
func (f *Follower) takeGrants(u agent.Update) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.grantsRev = u.Revision
|
||||
if u.Deleted {
|
||||
f.grants = nil
|
||||
f.keep(nil, 0)
|
||||
return
|
||||
}
|
||||
g, err := delivery.DecodeGrantsUserList(u.Value)
|
||||
if err != nil {
|
||||
f.grants = nil
|
||||
f.keep(nil, 0)
|
||||
f.sayOnceLocked("the grants part's user-list key cannot be read, so the controller's list is followed: " + err.Error())
|
||||
return
|
||||
}
|
||||
f.grants = &g
|
||||
f.keep(u.Value, u.Revision)
|
||||
}
|
||||
|
||||
// keep keeps the key last heard on the machine (Env.Keep), so a follower that cannot watch it follows what it said.
|
||||
func (f *Follower) keep(value []byte, rev uint64) {
|
||||
if f.env.Keep != nil {
|
||||
f.env.Keep(value, rev)
|
||||
}
|
||||
}
|
||||
|
||||
// Following is which list the follower takes now: the grants part's while its key says so, the controller's otherwise.
|
||||
func (f *Follower) Following() string {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.followingLocked()
|
||||
}
|
||||
|
||||
func (f *Follower) followingLocked() string {
|
||||
if f.grants != nil && f.grants.Follow == delivery.FollowGrants {
|
||||
return delivery.FollowGrants
|
||||
}
|
||||
return delivery.FollowController
|
||||
}
|
||||
|
||||
func (f *Follower) unwatchedNow() string {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.unwatched
|
||||
}
|
||||
|
||||
func (f *Follower) key() string {
|
||||
if f.env.Key == nil {
|
||||
return ""
|
||||
}
|
||||
return f.env.Key()
|
||||
}
|
||||
|
||||
// take keeps a composition key's change when it carries the bus's user list, and forgets a key deleted.
|
||||
func (f *Follower) take(u agent.Update) {
|
||||
node, module, _, isNode, err := delivery.ParseComposedKey(u.Key)
|
||||
@@ -192,6 +354,17 @@ func (f *Follower) take(u agent.Update) {
|
||||
// A source never goes back: a key deleted, or a newer one refused, leaves the list written as it is, since an older
|
||||
// list would take away accounts a newer one gave.
|
||||
func (f *Follower) settle(ctx context.Context, again bool) {
|
||||
f.mu.Lock()
|
||||
follow := f.followingLocked()
|
||||
if follow != f.following {
|
||||
// The switch turned: the other source is written, though neither changed.
|
||||
again = true
|
||||
}
|
||||
f.mu.Unlock()
|
||||
if follow == delivery.FollowGrants {
|
||||
f.settleGrants(ctx, again)
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
newest := f.cur
|
||||
for _, s := range f.srcs {
|
||||
@@ -207,7 +380,8 @@ func (f *Follower) settle(ctx context.Context, again bool) {
|
||||
at := fmt.Sprintf("%s@%d", newest.key, newest.rev)
|
||||
f.mu.Unlock()
|
||||
|
||||
w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, At: f.env.Now().UTC()}
|
||||
w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, At: f.env.Now().UTC(),
|
||||
Follows: delivery.FollowController, Key: f.key(), Unwatched: f.unwatchedNow()}
|
||||
var changed bool
|
||||
body, err := f.env.Resolve(ctx, newest.resource)
|
||||
if err == nil {
|
||||
@@ -220,16 +394,27 @@ func (f *Follower) settle(ctx context.Context, again bool) {
|
||||
}
|
||||
}
|
||||
}
|
||||
var loaded []byte
|
||||
if err == nil {
|
||||
w.Digest, changed, err = f.env.Write(ctx, newest.module, newest.resource, body)
|
||||
if err == nil {
|
||||
loaded = body
|
||||
}
|
||||
} else if body != nil && f.env.OnDisk != nil {
|
||||
// Refused: the list on the machine stays, and it is the one the bus has loaded.
|
||||
loaded, _ = f.env.OnDisk(newest.resource)
|
||||
}
|
||||
if err != nil {
|
||||
w.Error = err.Error()
|
||||
}
|
||||
f.mu.Lock()
|
||||
prev := f.state
|
||||
f.mu.Unlock()
|
||||
w.Handed = f.hand(loaded, prev.Handed)
|
||||
f.mu.Lock()
|
||||
f.mod, f.state = newest.module, w
|
||||
f.written = err == nil
|
||||
f.following = delivery.FollowController
|
||||
f.mu.Unlock()
|
||||
if err != nil {
|
||||
f.sayOnce(fmt.Sprintf("the bus's user list of %s could not be written: %v", at, err))
|
||||
@@ -238,12 +423,185 @@ func (f *Follower) settle(ctx context.Context, again bool) {
|
||||
at, w.Users, short(w.Digest)))
|
||||
}
|
||||
// Reported when what it says changed: a look again that found the file as it was says nothing new.
|
||||
if f.env.Report != nil && (changed || prev.Source != w.Source || prev.Revision != w.Revision ||
|
||||
prev.Digest != w.Digest || prev.Error != w.Error) {
|
||||
if f.env.Report != nil && (changed || reportable(prev, w)) {
|
||||
f.env.Report(ctx, newest.module, w)
|
||||
}
|
||||
}
|
||||
|
||||
// reportable says a have's user list says something new: a look again that found the file as it was says nothing.
|
||||
func reportable(prev, w Written) bool {
|
||||
return prev.Source != w.Source || prev.Revision != w.Revision || prev.Digest != w.Digest || prev.Error != w.Error ||
|
||||
prev.Follows != w.Follows || prev.Key != w.Key || handedKey(prev.Handed) != handedKey(w.Handed) ||
|
||||
prev.Unwatched != w.Unwatched
|
||||
}
|
||||
|
||||
// settleGrants writes the grants part's list, the object its key names, into the file the controller's composition of
|
||||
// the bus module declares, made 0600 (ADR 0547 clause 7): when the key moved, when the switch turned, or when asked.
|
||||
func (f *Follower) settleGrants(ctx context.Context, again bool) {
|
||||
f.mu.Lock()
|
||||
g, rev := *f.grants, f.grantsRev
|
||||
file := f.cur
|
||||
for _, s := range f.srcs {
|
||||
if s.rev > file.rev {
|
||||
file = s
|
||||
}
|
||||
}
|
||||
if !again && f.following == delivery.FollowGrants && f.state.Revision == rev && f.written {
|
||||
f.mu.Unlock()
|
||||
return
|
||||
}
|
||||
f.mu.Unlock()
|
||||
|
||||
w := Written{Source: delivery.GrantsUserListSource, Revision: rev, Wanted: g.Digest, At: f.env.Now().UTC(),
|
||||
Follows: delivery.FollowGrants, Key: f.key(), Unwatched: f.unwatchedNow()}
|
||||
var err error
|
||||
var changed bool
|
||||
resource, rerr := mode0600(file.resource)
|
||||
switch {
|
||||
case file.key == "" || rerr != nil:
|
||||
err = fmt.Errorf("no composition of this node declares the bus's user-list file, so the grants part's list has nowhere to be written")
|
||||
case g.Digest == "":
|
||||
err = fmt.Errorf("the grants part's key names no list yet; the list on the machine is kept")
|
||||
case f.env.Fetch == nil || f.env.Open == nil:
|
||||
err = fmt.Errorf("this engine cannot read or open the grants part's list")
|
||||
case g.SealedTo != "" && w.Key != "" && g.SealedTo != delivery.SealingKeyID(w.Key):
|
||||
err = fmt.Errorf("the grants part's list is sealed to another key than this node's (%s); the list on the machine is kept until it seals to this one", g.SealedTo)
|
||||
}
|
||||
var body []byte
|
||||
if err == nil {
|
||||
var object []byte
|
||||
if object, err = f.env.Fetch(ctx, g.Object); err == nil {
|
||||
body, err = delivery.OpenUserList(object, g.Digest, f.env.Open)
|
||||
}
|
||||
if err != nil {
|
||||
err = fmt.Errorf("the grants part's list %s could not be had: %w", short(g.Digest), err)
|
||||
}
|
||||
}
|
||||
if err == nil {
|
||||
w.Users = Users(body)
|
||||
if err = f.guard(resource, body); err != nil {
|
||||
if disk, derr := f.env.OnDisk(resource); derr == nil && disk != nil {
|
||||
w.Digest = Digest(disk)
|
||||
}
|
||||
}
|
||||
}
|
||||
if err == nil {
|
||||
w.Digest, changed, err = f.env.Write(ctx, file.module, resource, body)
|
||||
} else if w.Digest == "" && file.key != "" && f.env.OnDisk != nil {
|
||||
if disk, derr := f.env.OnDisk(resource); derr == nil && disk != nil {
|
||||
w.Digest = Digest(disk)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
w.Error = err.Error()
|
||||
}
|
||||
f.mu.Lock()
|
||||
prev := f.state
|
||||
// The grants part's list is not handed to the vault: the handing is of what the controller's list loaded (clause 9).
|
||||
w.Handed = prev.Handed
|
||||
f.mod, f.state = file.module, w
|
||||
f.written = err == nil
|
||||
f.following = delivery.FollowGrants
|
||||
f.mu.Unlock()
|
||||
at := fmt.Sprintf("%s@%d", delivery.GrantsUserListSource, rev)
|
||||
if err != nil {
|
||||
f.sayOnce(fmt.Sprintf("the bus's user list of %s could not be written: %v", at, err))
|
||||
} else if changed {
|
||||
f.env.Say(fmt.Sprintf("the bus's user list from the grants part (%s, revision %d) is written alone (%d users, %s); the bus reloads it itself",
|
||||
at, g.Revision, w.Users, short(w.Digest)))
|
||||
}
|
||||
if f.env.Report != nil && file.module != "" && (changed || reportable(prev, w)) {
|
||||
f.env.Report(ctx, file.module, w)
|
||||
}
|
||||
}
|
||||
|
||||
// Hold is c as the apply may write it (C3 of the switch's review): while the grants part's list is followed, the bus
|
||||
// module's composition from the controller carries the controller's list, so its user-list resource holds the list on the
|
||||
// machine instead (made 0600), or is left out when the machine holds none. A list on the machine that cannot be read
|
||||
// is an error: left out, the apply would remove the file as an orphan and the bus would refuse every connection at its
|
||||
// next reload, so the bus module's apply waits instead. Any other composition, or the controller's list followed, is c
|
||||
// unchanged.
|
||||
func (f *Follower) Hold(c delivery.Composed) (delivery.Composed, error) {
|
||||
if f.Following() != delivery.FollowGrants {
|
||||
return c, nil
|
||||
}
|
||||
out := c
|
||||
out.Resources = nil
|
||||
for _, raw := range c.Resources {
|
||||
if r, _, _, ok := userList([]json.RawMessage{raw}); ok {
|
||||
var held json.RawMessage
|
||||
body, err := f.env.OnDisk(r)
|
||||
if err != nil {
|
||||
return c, fmt.Errorf("the user list on the machine could not be read, so the apply waits and the file stays: %w", err)
|
||||
}
|
||||
if body != nil {
|
||||
var m map[string]any
|
||||
if json.Unmarshal(r, &m) == nil {
|
||||
delete(m, "digest")
|
||||
delete(m, "size")
|
||||
m["content"], m["mode"] = string(body), "0600"
|
||||
held, _ = json.Marshal(m)
|
||||
}
|
||||
}
|
||||
if held != nil {
|
||||
out.Resources = append(out.Resources, held)
|
||||
}
|
||||
continue
|
||||
}
|
||||
out.Resources = append(out.Resources, raw)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// mode0600 is a user-list resource made mode 0600, whatever its composition said (ADR 0547 clause 7).
|
||||
func mode0600(raw json.RawMessage) (json.RawMessage, error) {
|
||||
if len(raw) == 0 {
|
||||
return nil, fmt.Errorf("no resource")
|
||||
}
|
||||
var r map[string]any
|
||||
if err := json.Unmarshal(raw, &r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r["mode"] = "0600"
|
||||
return json.Marshal(r)
|
||||
}
|
||||
|
||||
// hand is the handing of the list on the machine to the vault: the one before when it is of the same list and key, a
|
||||
// new one sealed now otherwise, and none while the vault's key is not known or the list holds no hash.
|
||||
func (f *Follower) hand(loaded []byte, before *delivery.AccountsHanded) *delivery.AccountsHanded {
|
||||
if f.env.VaultKey == nil || loaded == nil {
|
||||
return before
|
||||
}
|
||||
key := f.env.VaultKey()
|
||||
if key == "" {
|
||||
return before
|
||||
}
|
||||
from := Digest(loaded)
|
||||
if before != nil && before.From == from && before.To == delivery.KeyFingerprint(key) {
|
||||
return before
|
||||
}
|
||||
accounts := delivery.HandedAccountsOf(loaded)
|
||||
if len(accounts) == 0 {
|
||||
return nil
|
||||
}
|
||||
h, err := delivery.HandAccounts(accounts, key, from, f.env.Now())
|
||||
if err != nil {
|
||||
f.sayOnce(fmt.Sprintf("the loaded list's hashes are not handed to the vault: %v", err))
|
||||
return before
|
||||
}
|
||||
f.env.Say(fmt.Sprintf("the hashes of the %d users of the bus's loaded list (%s) are handed to the vault, sealed to "+
|
||||
"its key %s, in the bus module's have", h.Users, short(from), short(h.To)))
|
||||
return &h
|
||||
}
|
||||
|
||||
// handedKey is what a handing says that a report compares: its list and key.
|
||||
func handedKey(h *delivery.AccountsHanded) string {
|
||||
if h == nil {
|
||||
return ""
|
||||
}
|
||||
return h.From + " " + h.To
|
||||
}
|
||||
|
||||
// ControllerUser is the controller's own user in the list: a list without it would lock the controller out of the bus.
|
||||
const ControllerUser = "controller"
|
||||
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/seal"
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
)
|
||||
|
||||
@@ -29,21 +30,48 @@ var _, strangerKey = key()
|
||||
|
||||
func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) }
|
||||
|
||||
// feed is a watch the test writes to: the keys there before the watch, Caught, then each put.
|
||||
// feed is a watch the test writes to: the keys there before the watch, Caught, then each put. watching is closed once
|
||||
// the watch began, so a put after it is never handed before the keys there before it.
|
||||
type feed struct {
|
||||
ch chan agent.Update
|
||||
rev uint64
|
||||
first []agent.Update
|
||||
// watching is closed once the watch caught up, so a later put is not read before the keys there before it.
|
||||
watching chan struct{}
|
||||
once sync.Once
|
||||
// err refuses the watch, as the bus refuses a watch it grants nothing of.
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *feed) Watch(context.Context, string) (<-chan agent.Update, error) {
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
for _, u := range f.first {
|
||||
f.ch <- u
|
||||
}
|
||||
f.ch <- agent.Update{Caught: true}
|
||||
f.once.Do(func() {
|
||||
if f.watching != nil {
|
||||
close(f.watching)
|
||||
}
|
||||
})
|
||||
return f.ch, nil
|
||||
}
|
||||
|
||||
// hashed is a list whose users carry a hash of bcrypt's form, as the composer writes them.
|
||||
func hashed(users ...string) string {
|
||||
var b strings.Builder
|
||||
b.WriteString("authorization {\n users = [\n")
|
||||
for _, u := range users {
|
||||
b.WriteString(` { user: "` + u + `", password: "$2a$11$` + strings.Repeat("h", 53) + `", permissions: {` + "\n")
|
||||
}
|
||||
b.WriteString(" ]\n}\n")
|
||||
return b.String()
|
||||
}
|
||||
|
||||
var vault, _ = seal.GenerateKeyPair()
|
||||
|
||||
func list(users ...string) string {
|
||||
var b strings.Builder
|
||||
b.WriteString("authorization {\n users = [\n")
|
||||
@@ -97,6 +125,15 @@ func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey, p
|
||||
return k, raw
|
||||
}
|
||||
|
||||
// gput is the grants part's key as the bus node hears it at the watch's start (ADR 0547 clauses 7, 8): which list to
|
||||
// follow, and the list sealed to the node's key; one sealed to another key, or one that does not open to the key's digest.
|
||||
type gput struct {
|
||||
follow, content string
|
||||
otherKey, badDigest bool
|
||||
// after puts it once the grants part's list was written, as a hand-back comes after a take-over.
|
||||
after bool
|
||||
}
|
||||
|
||||
type put struct {
|
||||
module, fp, content string
|
||||
stranger, deleted bool
|
||||
@@ -118,6 +155,18 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
wantUsers int
|
||||
wantErr string
|
||||
wantRefused bool
|
||||
grants []gput
|
||||
// grantsRefused refuses the watch of the grants part's key; kept is the key last heard, kept on the machine;
|
||||
// wantUnwatched says the have then says why it is not watched.
|
||||
grantsRefused bool
|
||||
kept *gput
|
||||
wantUnwatched bool
|
||||
// wantSource and wantFollows are what the last have says it took the list from and follows.
|
||||
wantSource, wantFollows string
|
||||
// vault gives the follower the vault's key; wantHanded is how many users it then hands from the list on the
|
||||
// machine (ADR 0547 clause 9).
|
||||
vault bool
|
||||
wantHanded int
|
||||
}{
|
||||
{name: "the list there is written once", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
||||
wantWrites: []string{two}, wantUsers: 3},
|
||||
@@ -151,17 +200,106 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
onDisk: five, wantUsers: 2, wantErr: "drops more than half", wantRefused: true},
|
||||
{name: "a list that drops half is written", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
||||
onDisk: list("controller", "a", "b", "c", "d", "e"), wantWrites: []string{two}, wantUsers: 3},
|
||||
{name: "the loaded list's hashes are handed sealed to the vault's key",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: hashed("controller", "node.ace")}}, vault: true,
|
||||
wantWrites: []string{hashed("controller", "node.ace")}, wantUsers: 2, wantHanded: 2},
|
||||
{name: "a refused list hands the list on the machine",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: list("a", "b")}}, vault: true,
|
||||
onDisk: hashed("controller", "a", "b"), wantUsers: 2, wantErr: "holds no user", wantRefused: true, wantHanded: 3},
|
||||
{name: "a failed write is said and kept", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, failWrite: true,
|
||||
wantWrites: []string{two}, wantUsers: 3, wantErr: "disk full"},
|
||||
// ADR 0547 clauses 7, 8: never both lists; the grants part's from take-over, the controller's otherwise.
|
||||
{name: "from take-over the grants part's list is written into the same file",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three}},
|
||||
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
||||
{name: "a hand-back writes the newest controller composition once",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}},
|
||||
grants: []gput{{follow: delivery.FollowGrants, content: three}, {follow: delivery.FollowController, content: three, after: true}},
|
||||
wantWrites: []string{three, two}, wantUsers: 3, wantSource: "novox.nats." + fpOf("a")[len(delivery.SourcePrefix2):], wantFollows: delivery.FollowController},
|
||||
{name: "a refused watch with no key kept follows the controller's list and says it is not watched",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grantsRefused: true, wantUnwatched: true,
|
||||
wantWrites: []string{two}, wantUsers: 3, wantSource: "novox.nats." + fpOf("a")[len(delivery.SourcePrefix2):], wantFollows: delivery.FollowController},
|
||||
{name: "a refused watch keeps the follow last heard, never the controller's list",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grantsRefused: true, wantUnwatched: true,
|
||||
kept: &gput{follow: delivery.FollowGrants, content: three},
|
||||
wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants},
|
||||
{name: "in shadow the grants part's list is not taken",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowController, content: three}},
|
||||
wantWrites: []string{two}, wantUsers: 3, wantSource: "novox.nats." + fpOf("a")[len(delivery.SourcePrefix2):], wantFollows: delivery.FollowController},
|
||||
{name: "a grants list that does not open to its digest is refused and the file kept",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three, badDigest: true}},
|
||||
onDisk: two, wantErr: "could not be had", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
||||
{name: "a grants list sealed to another key is refused and the file kept",
|
||||
before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three, otherKey: true}},
|
||||
onDisk: two, wantErr: "sealed to another key", wantRefused: true, wantSource: delivery.GrantsUserListSource},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
f := &feed{ch: make(chan agent.Update, 64)}
|
||||
f := &feed{ch: make(chan agent.Update, 64), watching: make(chan struct{})}
|
||||
objects := map[string]string{}
|
||||
var mu sync.Mutex
|
||||
var writes []string
|
||||
var reports []Written
|
||||
reported := make(chan struct{}, 64)
|
||||
fl := New(Env{Node: node, Composed: f,
|
||||
gf := &feed{ch: make(chan agent.Update, 64)}
|
||||
if c.grantsRefused {
|
||||
gf.err = errors.New("Permissions Violation for Publish to $JS.API.STREAM.INFO.KV_mesh-grants_user-list")
|
||||
}
|
||||
var grantsAfter []agent.Update
|
||||
var keptRaw []byte
|
||||
sealed := map[string][]byte{}
|
||||
nodeKey, err := seal.GenerateKeyPair()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gs := c.grants
|
||||
if c.kept != nil {
|
||||
gs = append(append([]gput(nil), gs...), *c.kept)
|
||||
}
|
||||
for i, g := range gs {
|
||||
to := nodeKey.PublicKey
|
||||
if g.otherKey {
|
||||
other, _ := seal.GenerateKeyPair()
|
||||
to = other.PublicKey
|
||||
}
|
||||
digest := delivery.UserListDigest([]byte(g.content))
|
||||
if g.badDigest {
|
||||
digest = delivery.UserListDigest([]byte("another list"))
|
||||
}
|
||||
object, err := delivery.SealUserList([]byte(g.content), to)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sealed[delivery.UserListObjectName(digest)] = object
|
||||
raw, err := delivery.EncodeGrantsUserList(delivery.GrantsUserList{Follow: g.follow, Digest: digest,
|
||||
Object: delivery.UserListObjectName(digest), SealedTo: delivery.SealingKeyID(to), Node: node, Revision: uint64(i + 1)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
u := agent.Update{Key: delivery.GrantsUserListKey, Value: raw, Revision: uint64(i + 1)}
|
||||
switch {
|
||||
case c.kept != nil && i == len(gs)-1:
|
||||
keptRaw = raw
|
||||
case g.after:
|
||||
grantsAfter = append(grantsAfter, u)
|
||||
default:
|
||||
gf.first = append(gf.first, u)
|
||||
}
|
||||
}
|
||||
fl := New(Env{Node: node, Composed: f, Grants: gf,
|
||||
Kept: func() ([]byte, uint64) { return keptRaw, 9 },
|
||||
Keep: func([]byte, uint64) {},
|
||||
Backoff: func(int) time.Duration { return time.Hour },
|
||||
Fetch: func(_ context.Context, name string) ([]byte, error) {
|
||||
if o, ok := sealed[name]; ok {
|
||||
return o, nil
|
||||
}
|
||||
return nil, errors.New("no object " + name)
|
||||
},
|
||||
Open: func(b seal.Box) ([]byte, error) {
|
||||
out, err := seal.Open(b, nodeKey.PrivateKey)
|
||||
return []byte(out), err
|
||||
},
|
||||
Key: func() string { return nodeKey.PublicKey },
|
||||
Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub} },
|
||||
Resolve: func(_ context.Context, r json.RawMessage) ([]byte, error) {
|
||||
var res struct{ Content, Digest string }
|
||||
@@ -197,6 +335,12 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
reported <- struct{}{}
|
||||
},
|
||||
Now: func() time.Time { return time.Date(2026, 10, 25, 21, 50, 0, 0, time.UTC) },
|
||||
VaultKey: func() string {
|
||||
if c.vault {
|
||||
return vault.PublicKey
|
||||
}
|
||||
return ""
|
||||
},
|
||||
})
|
||||
update := func(p put) agent.Update {
|
||||
f.rev++
|
||||
@@ -220,9 +364,31 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- fl.Run(ctx) }()
|
||||
<-f.watching
|
||||
// A later put after the first list is written: the first write waits for the grants part's key to catch up.
|
||||
if len(c.after) > 0 {
|
||||
for until := time.Now().Add(5 * time.Second); time.Now().Before(until); time.Sleep(5 * time.Millisecond) {
|
||||
mu.Lock()
|
||||
n := len(writes)
|
||||
mu.Unlock()
|
||||
if n > 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, p := range c.after {
|
||||
f.ch <- update(p)
|
||||
}
|
||||
if len(grantsAfter) > 0 {
|
||||
for until := time.Now().Add(5 * time.Second); ; time.Sleep(5 * time.Millisecond) {
|
||||
if _, st := fl.State(); st.Source == delivery.GrantsUserListSource || time.Now().After(until) {
|
||||
break
|
||||
}
|
||||
}
|
||||
for _, u := range grantsAfter {
|
||||
gf.ch <- u
|
||||
}
|
||||
}
|
||||
if c.again {
|
||||
// After the first write is reported, so the look again is not folded into it.
|
||||
select {
|
||||
@@ -252,6 +418,14 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
t.Fatal("a refused list was not reported")
|
||||
}
|
||||
}
|
||||
if c.wantSource != "" {
|
||||
for until := time.Now().Add(5 * time.Second); ; {
|
||||
if _, st := fl.State(); st.Source == c.wantSource && (c.wantErr == "" || st.Error != "") || time.Now().After(until) {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond) // nothing more is written
|
||||
cancel()
|
||||
<-done
|
||||
@@ -261,6 +435,54 @@ func TestTheBusNodeWritesTheNewestSignedUserListAlone(t *testing.T) {
|
||||
t.Fatalf("written %q, want %q", writes, c.wantWrites)
|
||||
}
|
||||
module, st := fl.State()
|
||||
if c.wantUnwatched != (st.Unwatched != "") {
|
||||
t.Fatalf("the have says unwatched %q", st.Unwatched)
|
||||
}
|
||||
// While the grants part's list is followed, an apply of the controller's composition writes not its list.
|
||||
if _, raw := comp(t, "nats", fpOf("a"), two, controllerKey, put{}); c.wantFollows != "" {
|
||||
cmp, err := delivery.DecodeComposed(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held, err := fl.Hold(cmp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _, _, hasList := userList(held.Resources)
|
||||
if want := c.wantFollows == delivery.FollowController; hasList != want || len(held.Resources) < 1 {
|
||||
t.Fatalf("an apply while following %s holds the controller's list: %v", c.wantFollows, hasList)
|
||||
}
|
||||
// A list on the machine that cannot be read stops the apply; it is never left out to be removed.
|
||||
if c.wantFollows == delivery.FollowGrants {
|
||||
readErr := New(Env{Node: node, Composed: f, OnDisk: func(json.RawMessage) ([]byte, error) {
|
||||
return nil, errors.New("permission denied")
|
||||
}})
|
||||
readErr.grants = &delivery.GrantsUserList{Follow: delivery.FollowGrants}
|
||||
if _, err := readErr.Hold(cmp); err == nil || !strings.Contains(err.Error(), "could not be read") {
|
||||
t.Fatalf("an unreadable list on the machine was held as %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if c.wantSource != "" && (st.Source != c.wantSource || st.Key != nodeKey.PublicKey ||
|
||||
(c.wantFollows != "" && st.Follows != c.wantFollows)) {
|
||||
t.Fatalf("the have says source %q follows %q key %t; want %q %q", st.Source, st.Follows, st.Key == nodeKey.PublicKey, c.wantSource, c.wantFollows)
|
||||
}
|
||||
switch {
|
||||
case c.wantHanded == 0 && st.Handed != nil:
|
||||
t.Fatalf("handed with no vault key or no hash: %+v", st.Handed)
|
||||
case c.wantHanded > 0:
|
||||
loaded := c.onDisk
|
||||
if !c.wantRefused {
|
||||
loaded = c.wantWrites[len(c.wantWrites)-1]
|
||||
}
|
||||
if st.Handed == nil || st.Handed.From != Digest([]byte(loaded)) {
|
||||
t.Fatalf("handed %+v, want from the list on the machine", st.Handed)
|
||||
}
|
||||
opened, err := delivery.OpenHanded(*st.Handed, vault.PublicKey, vault.PrivateKey)
|
||||
if err != nil || len(opened) != c.wantHanded || opened[0].User != "controller" {
|
||||
t.Fatalf("opened %+v, %v", opened, err)
|
||||
}
|
||||
}
|
||||
if c.wantRefused {
|
||||
if !strings.Contains(st.Error, c.wantErr) || st.Digest != Digest([]byte(c.onDisk)) || st.Users != c.wantUsers {
|
||||
t.Fatalf("refused list's state %+v", st)
|
||||
|
||||
Reference in New Issue
Block a user