295 lines
8.4 KiB
Go
295 lines
8.4 KiB
Go
package provider
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
provisions "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
)
|
|
|
|
// standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then
|
|
// every change, as the bus's key-value watch does.
|
|
type standIn struct {
|
|
mu sync.Mutex
|
|
buckets map[string]map[string][]byte
|
|
watchers map[string][]func(string, []byte)
|
|
}
|
|
|
|
func newStandIn() *standIn {
|
|
return &standIn{buckets: map[string]map[string][]byte{}, watchers: map[string][]func(string, []byte){}}
|
|
}
|
|
|
|
func (s *standIn) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
|
|
s.mu.Lock()
|
|
current := map[string][]byte{}
|
|
for k, v := range s.buckets[bucket] {
|
|
current[k] = v
|
|
}
|
|
s.watchers[bucket] = append(s.watchers[bucket], handle)
|
|
s.mu.Unlock()
|
|
for _, k := range sortedKeys(current) {
|
|
handle(k, current[k])
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *standIn) Put(_ context.Context, bucket, key string, value []byte) error {
|
|
s.mu.Lock()
|
|
if s.buckets[bucket] == nil {
|
|
s.buckets[bucket] = map[string][]byte{}
|
|
}
|
|
s.buckets[bucket][key] = append([]byte(nil), value...)
|
|
ws := slices.Clone(s.watchers[bucket])
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
w(key, value)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *standIn) Delete(_ context.Context, bucket, key string) error {
|
|
s.mu.Lock()
|
|
delete(s.buckets[bucket], key)
|
|
ws := slices.Clone(s.watchers[bucket])
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
w(key, nil)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteIf removes the key only while it still holds what the caller read.
|
|
func (s *standIn) DeleteIf(ctx context.Context, bucket, key string, held []byte) error {
|
|
s.mu.Lock()
|
|
cur := s.buckets[bucket][key]
|
|
s.mu.Unlock()
|
|
if !bytes.Equal(cur, held) {
|
|
return fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
return s.Delete(ctx, bucket, key)
|
|
}
|
|
|
|
func (s *standIn) get(bucket, key string) []byte {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.buckets[bucket][key]
|
|
}
|
|
|
|
func (s *standIn) keys(bucket string) []string {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return sortedKeys(s.buckets[bucket])
|
|
}
|
|
|
|
// everything is every value the bus holds, for a test that no secret is in a bucket.
|
|
func (s *standIn) everything() []byte {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var all []byte
|
|
for _, b := range s.buckets {
|
|
for k, v := range b {
|
|
all = append(all, k...)
|
|
all = append(all, v...)
|
|
}
|
|
}
|
|
return all
|
|
}
|
|
|
|
// vault is a stand-in vault: a value per name, or nothing yet.
|
|
type vault struct {
|
|
mu sync.Mutex
|
|
values map[string]string
|
|
asked []string
|
|
}
|
|
|
|
func (v *vault) Value(_ context.Context, name string) ([]byte, error) {
|
|
v.mu.Lock()
|
|
defer v.mu.Unlock()
|
|
v.asked = append(v.asked, name)
|
|
val, ok := v.values[name]
|
|
if !ok {
|
|
return nil, fmt.Errorf("the vault holds no %s yet", name)
|
|
}
|
|
return []byte(val), nil
|
|
}
|
|
|
|
// maker records what the provider's adapter was asked to make, with the credential it was handed.
|
|
type maker struct {
|
|
mu sync.Mutex
|
|
made map[string]string
|
|
// contributed is what each consumer's entry carried of its plan's contribution.
|
|
contributed map[string]map[string]any
|
|
// as is the identity each consumer's entry was made under.
|
|
as map[string]string
|
|
gone []string
|
|
fail error
|
|
calls int
|
|
}
|
|
|
|
func (m *maker) Make(_ context.Context, c Consumer) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.calls++
|
|
if m.fail != nil {
|
|
return m.fail
|
|
}
|
|
if m.made == nil {
|
|
m.made = map[string]string{}
|
|
}
|
|
m.made[c.Entry.Key()] = string(c.Credential)
|
|
if m.contributed == nil {
|
|
m.contributed = map[string]map[string]any{}
|
|
}
|
|
m.contributed[c.Entry.Key()] = c.Entry.Contributes
|
|
if m.as == nil {
|
|
m.as = map[string]string{}
|
|
}
|
|
m.as[c.Entry.Key()] = c.Entry.As
|
|
return nil
|
|
}
|
|
|
|
func (m *maker) Gone(_ context.Context, e Entry) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.gone = append(m.gone, e.Key())
|
|
return nil
|
|
}
|
|
|
|
// assign puts a consumer's plan with its provisions on the seat mesh-provisioner, each provision's key provided, as its
|
|
// holder writes them (novox/hq ADR 0570 clauses 1, 2): what a provider binds by.
|
|
func assign(t *testing.T, bus *standIn, node, module string, binds ...map[string]string) {
|
|
t.Helper()
|
|
var ps []catalogue.PlanProvision
|
|
for _, b := range binds {
|
|
raw, _ := json.Marshal(b)
|
|
var pp catalogue.PlanProvision
|
|
if err := json.Unmarshal(raw, &pp); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ps = append(ps, pp)
|
|
}
|
|
planned(t, bus, node, module, ps...)
|
|
}
|
|
|
|
// planned puts a consumer's plan of these provisions on the seat, each provision's key provided.
|
|
func planned(t *testing.T, bus *standIn, node, module string, ps ...catalogue.PlanProvision) {
|
|
t.Helper()
|
|
for _, pp := range ps {
|
|
pr := provisions.Provision{Provision: pp.Provision, Local: pp.Local, Consumer: provisions.End{Module: module, Node: node},
|
|
Provider: provisions.End{Module: pp.ProviderModule, Node: pp.Provider}, State: provisions.Provided}
|
|
key, err := pr.Key()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
value, err := provisions.Encode(pr)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = bus.Put(context.Background(), provisions.ProvisionsBucket, key, value)
|
|
}
|
|
raw, err := provisions.EncodePlan(provisions.Plan{Node: node, Module: module, Plan: catalogue.Plan{Provisions: ps}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := bus.Put(context.Background(), provisions.PlansBucket, node+"."+module, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// retire marks a consumer's provision retiring, with the operator's yes (novox/hq ADR 0570 clause 5): the one way a
|
|
// provider lets a consumer go.
|
|
func retire(t *testing.T, bus *standIn, node, module, provision, provider, providerModule string) {
|
|
t.Helper()
|
|
pr := provisions.Provision{Provision: provision, Consumer: provisions.End{Module: module, Node: node},
|
|
Provider: provisions.End{Module: providerModule, Node: provider}, State: provisions.Retiring, Why: module + " left " + node,
|
|
Warrant: &provisions.Warrant{Ask: "a-1", Option: "yes", By: "the operator"}}
|
|
key, _ := pr.Key()
|
|
raw, err := provisions.Encode(pr)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = bus.Delete(context.Background(), provisions.PlansBucket, node+"."+module)
|
|
_ = bus.Put(context.Background(), provisions.ProvisionsBucket, key, raw)
|
|
}
|
|
|
|
func bind(provision, local, provider, providerModule string) map[string]string {
|
|
b := map[string]string{"provision": provision, "provider": provider, "provider-module": providerModule}
|
|
if local != "" {
|
|
b["local"] = local
|
|
}
|
|
return b
|
|
}
|
|
|
|
func entryOf(t *testing.T, bus *standIn, key string) Entry {
|
|
t.Helper()
|
|
raw := bus.get(BucketOf("postgres"), key)
|
|
if raw == nil {
|
|
t.Fatalf("no entry %s in %s; it holds %v", key, BucketOf("postgres"), bus.keys(BucketOf("postgres")))
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(raw, &e); err != nil {
|
|
t.Fatalf("entry %s: %v", key, err)
|
|
}
|
|
return e
|
|
}
|
|
|
|
// postgres is a provider of postgres-database on novox with a login per consumer (ADR 0048, ADR 0158 §11), its
|
|
// served values derived from each consumer's identity (ADR 0202 §1).
|
|
func postgres(bus *standIn, v *vault, m *maker) *Provider {
|
|
return New(Options{
|
|
Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Gone: m.Gone,
|
|
Offers: map[string]Offer{
|
|
"postgres-database": {Credential: PerConsumer, Serves: map[string]any{
|
|
"host": "postgres.novox.mesh", "port": 5432, "database": "${consumer:as}"}},
|
|
},
|
|
Logf: func(string, ...any) {},
|
|
})
|
|
}
|
|
|
|
func waitFor(t *testing.T, ok func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
for !ok() {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("not within 3s")
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// refusedByTheStateGuard mirrors the key rule of the runtime's state guard (mesh-tools node-tools internal/bus
|
|
// state.go): a JSON key whose lowercased name, with `-`, `_` and spaces taken out, ends in one of these is refused.
|
|
func refusedByTheStateGuard(key string) bool {
|
|
k := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(key))
|
|
for _, end := range []string{"password", "passwd", "secret", "token", "credential", "credentials",
|
|
"authorization", "apikey", "privatekey", "accesskey", "cookie"} {
|
|
if strings.HasSuffix(k, end) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func keysOf(v any, into *[]string) {
|
|
switch x := v.(type) {
|
|
case map[string]any:
|
|
for k, c := range x {
|
|
*into = append(*into, k)
|
|
keysOf(c, into)
|
|
}
|
|
case []any:
|
|
for _, c := range x {
|
|
keysOf(c, into)
|
|
}
|
|
}
|
|
}
|